You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@dolphinscheduler.apache.org by ch...@apache.org on 2024/04/03 05:56:47 UTC
(dolphinscheduler) branch dev updated: fix dependent task logic (#15795)
This is an automated email from the ASF dual-hosted git repository.
chufenggao pushed a commit to branch dev
in repository https://gitbox.apache.org/repos/asf/dolphinscheduler.git
The following commit(s) were added to refs/heads/dev by this push:
new c0435e5bd8 fix dependent task logic (#15795)
c0435e5bd8 is described below
commit c0435e5bd88ab8712942cfecf9e1fefa0d886773
Author: Evan Sun <ab...@gmail.com>
AuthorDate: Wed Apr 3 13:56:41 2024 +0800
fix dependent task logic (#15795)
Co-authored-by: abzymeinsjtu <ab...@B-54Q8MD6R-0244.local>
---
.../dolphinscheduler/dao/mapper/ProcessInstanceMapper.java | 6 ++++--
.../dolphinscheduler/dao/repository/ProcessInstanceDao.java | 6 ++++--
.../dao/repository/impl/ProcessInstanceDaoImpl.java | 10 +++++++---
.../dolphinscheduler/dao/mapper/ProcessInstanceMapper.xml | 9 +++++++--
.../dolphinscheduler/dao/mapper/ProcessInstanceMapperTest.java | 3 ++-
.../dolphinscheduler/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);