This is an automated email from the ASF dual-hosted git repository.
SbloodyS 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 673b361582 [Improvement-18249][DAO] Route WorkflowTaskRelationMapper
access through WorkflowTaskRelationDao (#18260)
673b361582 is described below
commit 673b3615824c7e01d7435163e72509706b2e1605
Author: Wenjun Ruan <[email protected]>
AuthorDate: Sun May 17 22:28:02 2026 +0800
[Improvement-18249][DAO] Route WorkflowTaskRelationMapper access through
WorkflowTaskRelationDao (#18260)
---
.../api/service/impl/ExecutorServiceImpl.java | 6 +-
.../service/impl/TaskDefinitionServiceImpl.java | 26 +++---
.../impl/WorkflowDefinitionServiceImpl.java | 14 ++--
.../impl/WorkflowTaskRelationServiceImpl.java | 8 +-
.../api/service/TaskDefinitionServiceImplTest.java | 10 ++-
.../api/service/WorkflowDefinitionServiceTest.java | 10 +--
.../dao/repository/WorkflowTaskRelationDao.java | 47 +++++++++++
.../impl/WorkflowTaskRelationDaoImpl.java | 93 ++++++++++++++++++++++
8 files changed, 178 insertions(+), 36 deletions(-)
diff --git
a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/ExecutorServiceImpl.java
b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/ExecutorServiceImpl.java
index 458b5eea10..70d3f084ec 100644
---
a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/ExecutorServiceImpl.java
+++
b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/ExecutorServiceImpl.java
@@ -53,9 +53,9 @@ import
org.apache.dolphinscheduler.dao.entity.WorkflowTaskRelation;
import org.apache.dolphinscheduler.dao.mapper.TaskDefinitionLogMapper;
import org.apache.dolphinscheduler.dao.mapper.TaskDefinitionMapper;
import org.apache.dolphinscheduler.dao.mapper.TaskGroupQueueMapper;
-import org.apache.dolphinscheduler.dao.mapper.WorkflowTaskRelationMapper;
import org.apache.dolphinscheduler.dao.repository.WorkflowDefinitionDao;
import org.apache.dolphinscheduler.dao.repository.WorkflowInstanceDao;
+import org.apache.dolphinscheduler.dao.repository.WorkflowTaskRelationDao;
import org.apache.dolphinscheduler.plugin.task.api.utils.TaskTypeUtils;
import org.apache.dolphinscheduler.service.command.CommandService;
import org.apache.dolphinscheduler.service.process.ProcessService;
@@ -102,7 +102,7 @@ public class ExecutorServiceImpl extends BaseServiceImpl
implements ExecutorServ
private TaskDefinitionMapper taskDefinitionMapper;
@Autowired
- private WorkflowTaskRelationMapper workflowTaskRelationMapper;
+ private WorkflowTaskRelationDao workflowTaskRelationDao;
@Autowired
private TaskGroupQueueMapper taskGroupQueueMapper;
@@ -197,7 +197,7 @@ public class ExecutorServiceImpl extends BaseServiceImpl
implements ExecutorServ
public boolean checkSubWorkflowDefinitionValid(WorkflowDefinition
workflowDefinition) {
// query all sub workflows under the current workflow
List<WorkflowTaskRelation> workflowTaskRelations =
-
workflowTaskRelationMapper.queryDownstreamByWorkflowDefinitionCode(workflowDefinition.getCode());
+
workflowTaskRelationDao.queryDownstreamByWorkflowDefinitionCode(workflowDefinition.getCode());
if (workflowTaskRelations.isEmpty()) {
return true;
}
diff --git
a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/TaskDefinitionServiceImpl.java
b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/TaskDefinitionServiceImpl.java
index f5fb12dc2a..be2764c873 100644
---
a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/TaskDefinitionServiceImpl.java
+++
b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/TaskDefinitionServiceImpl.java
@@ -51,10 +51,10 @@ import
org.apache.dolphinscheduler.dao.entity.WorkflowTaskRelationLog;
import org.apache.dolphinscheduler.dao.mapper.TaskDefinitionLogMapper;
import org.apache.dolphinscheduler.dao.mapper.TaskDefinitionMapper;
import org.apache.dolphinscheduler.dao.mapper.WorkflowDefinitionLogMapper;
-import org.apache.dolphinscheduler.dao.mapper.WorkflowTaskRelationMapper;
import org.apache.dolphinscheduler.dao.repository.ProjectDao;
import org.apache.dolphinscheduler.dao.repository.TaskDefinitionDao;
import org.apache.dolphinscheduler.dao.repository.WorkflowDefinitionDao;
+import org.apache.dolphinscheduler.dao.repository.WorkflowTaskRelationDao;
import org.apache.dolphinscheduler.dao.repository.WorkflowTaskRelationLogDao;
import org.apache.dolphinscheduler.service.process.ProcessService;
@@ -104,7 +104,7 @@ public class TaskDefinitionServiceImpl extends
BaseServiceImpl implements TaskDe
private TaskDefinitionLogMapper taskDefinitionLogMapper;
@Autowired
- private WorkflowTaskRelationMapper workflowTaskRelationMapper;
+ private WorkflowTaskRelationDao workflowTaskRelationDao;
@Autowired
private WorkflowTaskRelationLogDao workflowTaskRelationLogDao;
@@ -270,7 +270,7 @@ public class TaskDefinitionServiceImpl extends
BaseServiceImpl implements TaskDe
"Update task definition and definitionLog complete,
projectCode:{}, taskDefinitionCode:{}, newTaskVersion:{}.",
projectCode, taskCode, taskDefinitionToUpdate.getVersion());
// update workflow task relation
- List<WorkflowTaskRelation> workflowTaskRelations =
workflowTaskRelationMapper
+ List<WorkflowTaskRelation> workflowTaskRelations =
workflowTaskRelationDao
.queryWorkflowTaskRelationByTaskCodeAndTaskVersion(taskDefinitionToUpdate.getCode(),
taskDefinition.getVersion());
if (CollectionUtils.isNotEmpty(workflowTaskRelations)) {
@@ -290,9 +290,7 @@ public class TaskDefinitionServiceImpl extends
BaseServiceImpl implements TaskDe
workflowTaskRelation.setPostTaskVersion(version);
}
workflowTaskRelation.setWorkflowDefinitionVersion(workflowDefinitionVersion);
- int updateWorkflowDefinitionVersionCount =
-
workflowTaskRelationMapper.updateWorkflowTaskRelationTaskVersion(workflowTaskRelation);
- if (updateWorkflowDefinitionVersionCount != 1) {
+ if
(!workflowTaskRelationDao.updateWorkflowTaskRelationTaskVersion(workflowTaskRelation))
{
log.error("batch update workflow task relation error,
projectCode:{}, taskDefinitionCode:{}.",
projectCode, taskCode);
throw new
ServiceException(Status.WORKFLOW_TASK_RELATION_BATCH_UPDATE_ERROR);
@@ -344,7 +342,7 @@ public class TaskDefinitionServiceImpl extends
BaseServiceImpl implements TaskDe
TaskDefinitionLog taskDefinitionToUpdate =
updateTask(loginUser, projectCode, taskCode,
taskDefinitionJsonObj);
List<WorkflowTaskRelation> upstreamTaskRelations =
- workflowTaskRelationMapper.queryUpstreamByCode(projectCode,
taskCode);
+ workflowTaskRelationDao.queryUpstreamByCode(projectCode,
taskCode);
Set<Long> upstreamCodeSet =
upstreamTaskRelations.stream().map(WorkflowTaskRelation::getPreTaskCode).collect(Collectors.toSet());
Set<Long> upstreamTaskCodes = Collections.emptySet();
@@ -374,7 +372,7 @@ public class TaskDefinitionServiceImpl extends
BaseServiceImpl implements TaskDe
if (MapUtils.isNotEmpty(queryUpStreamTaskCodeMap)) {
WorkflowTaskRelation taskRelation = upstreamTaskRelations.get(0);
List<WorkflowTaskRelation> workflowTaskRelations =
-
workflowTaskRelationMapper.queryByWorkflowDefinitionCode(taskRelation.getWorkflowDefinitionCode());
+
workflowTaskRelationDao.queryByWorkflowDefinitionCode(taskRelation.getWorkflowDefinitionCode());
// set upstream code list
updateUpstreamTask(new
HashSet<>(queryUpStreamTaskCodeMap.keySet()),
@@ -413,7 +411,7 @@ public class TaskDefinitionServiceImpl extends
BaseServiceImpl implements TaskDe
private void updateUpstreamTask(Set<Long> allPreTaskCodeSet, long
taskCode, long projectCode,
long workflowDefinitionCode, User
loginUser) {
// query all workflow task relation
- List<WorkflowTaskRelation> hadWorkflowTaskRelationList =
workflowTaskRelationMapper
+ List<WorkflowTaskRelation> hadWorkflowTaskRelationList =
workflowTaskRelationDao
.queryUpstreamByCode(projectCode, taskCode);
// remove pre
Set<Long> removePreTaskSet = new HashSet<>();
@@ -470,11 +468,11 @@ public class TaskDefinitionServiceImpl extends
BaseServiceImpl implements TaskDe
int log = 0;
// insert workflow task relation table data
if (CollectionUtils.isNotEmpty(addPreTaskList)) {
- insert = workflowTaskRelationMapper.batchInsert(addPreTaskList);
+ insert = workflowTaskRelationDao.batchInsert(addPreTaskList);
}
if (CollectionUtils.isNotEmpty(removePreTaskList)) {
for (WorkflowTaskRelation workflowTaskRelation :
removePreTaskList) {
- remove +=
workflowTaskRelationMapper.updateById(workflowTaskRelation);
+ remove +=
workflowTaskRelationDao.updateById(workflowTaskRelation) ? 1 : 0;
}
}
if (CollectionUtils.isNotEmpty(workflowTaskRelationLogList)) {
@@ -533,14 +531,14 @@ public class TaskDefinitionServiceImpl extends
BaseServiceImpl implements TaskDe
throw new
ServiceException(Status.SWITCH_TASK_DEFINITION_VERSION_ERROR);
}
List<WorkflowTaskRelation> taskRelationList =
- workflowTaskRelationMapper.queryUpstreamByCode(projectCode,
taskCode);
+ workflowTaskRelationDao.queryUpstreamByCode(projectCode,
taskCode);
if (CollectionUtils.isNotEmpty(taskRelationList)) {
log.info(
"Task definition has upstream tasks, start handle them
after switch task, taskDefinitionCode:{}.",
taskCode);
long workflowDefinitionCode =
taskRelationList.get(0).getWorkflowDefinitionCode();
List<WorkflowTaskRelation> workflowTaskRelations =
-
workflowTaskRelationMapper.queryByWorkflowDefinitionCode(workflowDefinitionCode);
+
workflowTaskRelationDao.queryByWorkflowDefinitionCode(workflowDefinitionCode);
updateDag(loginUser, workflowDefinitionCode, workflowTaskRelations,
Lists.newArrayList(taskDefinitionUpdate));
} else {
@@ -613,7 +611,7 @@ public class TaskDefinitionServiceImpl extends
BaseServiceImpl implements TaskDe
log.error("Task definition does not exist,
taskDefinitionCode:{}.", taskCode);
throw new ServiceException(Status.TASK_DEFINE_NOT_EXIST,
String.valueOf(taskCode));
}
- List<WorkflowTaskRelation> taskRelationList =
workflowTaskRelationMapper
+ List<WorkflowTaskRelation> taskRelationList = workflowTaskRelationDao
.queryByCode(projectCode, 0, 0, taskCode);
if (CollectionUtils.isNotEmpty(taskRelationList)) {
taskRelationList = taskRelationList.stream()
diff --git
a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/WorkflowDefinitionServiceImpl.java
b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/WorkflowDefinitionServiceImpl.java
index 54a165be05..e883fe99c3 100644
---
a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/WorkflowDefinitionServiceImpl.java
+++
b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/WorkflowDefinitionServiceImpl.java
@@ -81,7 +81,6 @@ import
org.apache.dolphinscheduler.dao.mapper.TaskDefinitionLogMapper;
import org.apache.dolphinscheduler.dao.mapper.TaskDefinitionMapper;
import org.apache.dolphinscheduler.dao.mapper.WorkflowDefinitionLogMapper;
import org.apache.dolphinscheduler.dao.mapper.WorkflowTaskRelationLogMapper;
-import org.apache.dolphinscheduler.dao.mapper.WorkflowTaskRelationMapper;
import org.apache.dolphinscheduler.dao.model.PageListingResult;
import org.apache.dolphinscheduler.dao.repository.ProjectDao;
import org.apache.dolphinscheduler.dao.repository.ScheduleDao;
@@ -90,6 +89,7 @@ import
org.apache.dolphinscheduler.dao.repository.TaskInstanceDao;
import org.apache.dolphinscheduler.dao.repository.UserDao;
import org.apache.dolphinscheduler.dao.repository.WorkflowDefinitionDao;
import org.apache.dolphinscheduler.dao.repository.WorkflowDefinitionLogDao;
+import org.apache.dolphinscheduler.dao.repository.WorkflowTaskRelationDao;
import
org.apache.dolphinscheduler.plugin.task.api.model.ConditionDependentItem;
import
org.apache.dolphinscheduler.plugin.task.api.model.ConditionDependentTaskModel;
import org.apache.dolphinscheduler.plugin.task.api.model.DependentItem;
@@ -181,7 +181,7 @@ public class WorkflowDefinitionServiceImpl extends
BaseServiceImpl implements Wo
private TaskDefinitionLogDao taskDefinitionLogDao;
@Autowired
- private WorkflowTaskRelationMapper workflowTaskRelationMapper;
+ private WorkflowTaskRelationDao workflowTaskRelationDao;
@Autowired
private WorkflowTaskRelationLogMapper workflowTaskRelationLogMapper;
@@ -668,7 +668,7 @@ public class WorkflowDefinitionServiceImpl extends
BaseServiceImpl implements Wo
private void taskUsedInOtherTaskValid(WorkflowDefinition
workflowDefinition,
List<WorkflowTaskRelationLog>
taskRelationList) {
List<WorkflowTaskRelation> oldWorkflowTaskRelationList =
-
workflowTaskRelationMapper.queryByWorkflowDefinitionCode(workflowDefinition.getCode());
+
workflowTaskRelationDao.queryByWorkflowDefinitionCode(workflowDefinition.getCode());
Set<WorkflowTaskRelationLog> oldWorkflowTaskRelationSet =
oldWorkflowTaskRelationList.stream().map(WorkflowTaskRelationLog::new).collect(Collectors.toSet());
StringBuilder sb = new StringBuilder();
@@ -1325,7 +1325,7 @@ public class WorkflowDefinitionServiceImpl extends
BaseServiceImpl implements Wo
diffCode.forEach(code -> failedWorkflowList.add(code + "[null]"));
for (WorkflowDefinition workflowDefinition : workflowDefinitionList) {
List<WorkflowTaskRelation> workflowTaskRelations =
-
workflowTaskRelationMapper.queryByWorkflowDefinitionCode(workflowDefinition.getCode());
+
workflowTaskRelationDao.queryByWorkflowDefinitionCode(workflowDefinition.getCode());
List<WorkflowTaskRelationLog> taskRelationList =
workflowTaskRelations.stream().map(WorkflowTaskRelationLog::new).collect(Collectors.toList());
workflowDefinition.setProjectCode(targetProjectCode);
@@ -1586,7 +1586,7 @@ public class WorkflowDefinitionServiceImpl extends
BaseServiceImpl implements Wo
throw new
ServiceException(Status.SWITCH_WORKFLOW_DEFINITION_VERSION_ERROR);
}
- List<WorkflowTaskRelation> workflowTaskRelationList =
workflowTaskRelationMapper
+ List<WorkflowTaskRelation> workflowTaskRelationList =
workflowTaskRelationDao
.queryWorkflowTaskRelationsByWorkflowDefinitionCode(workflowDefinitionLog.getCode(),
workflowDefinitionLog.getVersion());
List<TaskCodeVersionDto> taskDefinitionList =
getTaskCodeVersionDtos(workflowTaskRelationList);
@@ -1806,7 +1806,7 @@ public class WorkflowDefinitionServiceImpl extends
BaseServiceImpl implements Wo
Set<Long> taskCodeSet = new TreeSet<>();
-
workflowTaskRelationMapper.queryByWorkflowDefinitionCode(workflowDefinition.getCode())
+
workflowTaskRelationDao.queryByWorkflowDefinitionCode(workflowDefinition.getCode())
.forEach(processTaskRelation -> {
if (processTaskRelation.getPreTaskCode() > 0) {
taskCodeSet.add(processTaskRelation.getPreTaskCode());
@@ -1836,7 +1836,7 @@ public class WorkflowDefinitionServiceImpl extends
BaseServiceImpl implements Wo
private void checkWorkflowDefinitionIsValidated(Long
workflowDefinitionCode) {
// todo: build dag check if the dag is validated
List<WorkflowTaskRelation> workflowTaskRelations =
-
workflowTaskRelationMapper.queryByWorkflowDefinitionCode(workflowDefinitionCode);
+
workflowTaskRelationDao.queryByWorkflowDefinitionCode(workflowDefinitionCode);
if (CollectionUtils.isEmpty(workflowTaskRelations)) {
throw new ServiceException(Status.WORKFLOW_DAG_IS_EMPTY);
}
diff --git
a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/WorkflowTaskRelationServiceImpl.java
b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/WorkflowTaskRelationServiceImpl.java
index 620a04eafe..760a5f02fa 100644
---
a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/WorkflowTaskRelationServiceImpl.java
+++
b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/WorkflowTaskRelationServiceImpl.java
@@ -19,7 +19,7 @@ package org.apache.dolphinscheduler.api.service.impl;
import org.apache.dolphinscheduler.api.service.WorkflowTaskRelationService;
import org.apache.dolphinscheduler.dao.entity.WorkflowTaskRelation;
-import org.apache.dolphinscheduler.dao.mapper.WorkflowTaskRelationMapper;
+import org.apache.dolphinscheduler.dao.repository.WorkflowTaskRelationDao;
import java.util.List;
@@ -33,18 +33,18 @@ import org.springframework.stereotype.Service;
public class WorkflowTaskRelationServiceImpl extends BaseServiceImpl
implements WorkflowTaskRelationService {
@Autowired
- private WorkflowTaskRelationMapper workflowTaskRelationMapper;
+ private WorkflowTaskRelationDao workflowTaskRelationDao;
@Override
public List<WorkflowTaskRelation> queryByWorkflowDefinitionCode(long
workflowDefinitionCode,
int
workflowDefinitionVersion) {
- return
workflowTaskRelationMapper.queryWorkflowTaskRelationsByWorkflowDefinitionCode(workflowDefinitionCode,
+ return
workflowTaskRelationDao.queryWorkflowTaskRelationsByWorkflowDefinitionCode(workflowDefinitionCode,
workflowDefinitionVersion);
}
@Override
public void deleteByWorkflowDefinitionCode(long workflowDefinitionCode,
int workflowDefinitionVersion) {
-
workflowTaskRelationMapper.deleteByWorkflowDefinitionCodeAndVersion(workflowDefinitionCode,
+
workflowTaskRelationDao.deleteByWorkflowDefinitionCodeAndVersion(workflowDefinitionCode,
workflowDefinitionVersion);
}
diff --git
a/dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/service/TaskDefinitionServiceImplTest.java
b/dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/service/TaskDefinitionServiceImplTest.java
index a82cf6e1a5..a8affea832 100644
---
a/dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/service/TaskDefinitionServiceImplTest.java
+++
b/dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/service/TaskDefinitionServiceImplTest.java
@@ -53,6 +53,7 @@ import
org.apache.dolphinscheduler.dao.mapper.WorkflowTaskRelationLogMapper;
import org.apache.dolphinscheduler.dao.mapper.WorkflowTaskRelationMapper;
import org.apache.dolphinscheduler.dao.repository.ProjectDao;
import org.apache.dolphinscheduler.dao.repository.WorkflowDefinitionDao;
+import org.apache.dolphinscheduler.dao.repository.WorkflowTaskRelationDao;
import org.apache.dolphinscheduler.dao.repository.WorkflowTaskRelationLogDao;
import org.apache.dolphinscheduler.plugin.task.api.TaskPluginManager;
import org.apache.dolphinscheduler.service.process.ProcessService;
@@ -110,6 +111,9 @@ public class TaskDefinitionServiceImplTest {
@Mock
private WorkflowTaskRelationMapper workflowTaskRelationMapper;
+ @Mock
+ private WorkflowTaskRelationDao workflowTaskRelationDao;
+
@Mock
private WorkflowDefinitionMapper workflowDefinitionMapper;
@@ -364,12 +368,12 @@ public class TaskDefinitionServiceImplTest {
when(taskDefinitionMapper.queryByCodeList(Mockito.anySet()))
.thenReturn(Arrays.asList(taskDefinition,
taskDefinitionSecond));
- when(workflowTaskRelationMapper.queryUpstreamByCode(PROJECT_CODE,
TASK_CODE))
+ when(workflowTaskRelationDao.queryUpstreamByCode(PROJECT_CODE,
TASK_CODE))
.thenReturn(getProcessTaskRelationListV2());
when(workflowDefinitionDao.queryByCode(PROCESS_DEFINITION_CODE))
.thenReturn(Optional.of(getProcessDefinition()));
-
when(workflowTaskRelationMapper.batchInsert(Mockito.anyList())).thenReturn(1);
-
when(workflowTaskRelationMapper.updateById(Mockito.any())).thenReturn(1);
+
when(workflowTaskRelationDao.batchInsert(Mockito.anyList())).thenReturn(1);
+
when(workflowTaskRelationDao.updateById(Mockito.any())).thenReturn(true);
when(workflowTaskRelationLogDao.batchInsert(Mockito.anyList())).thenReturn(2);
// success
Long updatedTaskCode =
taskDefinitionService.updateTaskWithUpstream(user, PROJECT_CODE, TASK_CODE,
diff --git
a/dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/service/WorkflowDefinitionServiceTest.java
b/dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/service/WorkflowDefinitionServiceTest.java
index 168d9876bc..b7654c5427 100644
---
a/dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/service/WorkflowDefinitionServiceTest.java
+++
b/dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/service/WorkflowDefinitionServiceTest.java
@@ -64,7 +64,6 @@ import
org.apache.dolphinscheduler.dao.mapper.TaskDefinitionLogMapper;
import org.apache.dolphinscheduler.dao.mapper.TaskDefinitionMapper;
import org.apache.dolphinscheduler.dao.mapper.WorkflowDefinitionLogMapper;
import org.apache.dolphinscheduler.dao.mapper.WorkflowTaskRelationLogMapper;
-import org.apache.dolphinscheduler.dao.mapper.WorkflowTaskRelationMapper;
import org.apache.dolphinscheduler.dao.model.PageListingResult;
import org.apache.dolphinscheduler.dao.repository.ProjectDao;
import org.apache.dolphinscheduler.dao.repository.ScheduleDao;
@@ -73,6 +72,7 @@ import
org.apache.dolphinscheduler.dao.repository.TaskInstanceDao;
import org.apache.dolphinscheduler.dao.repository.UserDao;
import org.apache.dolphinscheduler.dao.repository.WorkflowDefinitionDao;
import org.apache.dolphinscheduler.dao.repository.WorkflowDefinitionLogDao;
+import org.apache.dolphinscheduler.dao.repository.WorkflowTaskRelationDao;
import org.apache.dolphinscheduler.dao.utils.WorkerGroupUtils;
import
org.apache.dolphinscheduler.plugin.task.api.model.ConditionDependentItem;
import
org.apache.dolphinscheduler.plugin.task.api.model.ConditionDependentTaskModel;
@@ -150,7 +150,7 @@ public class WorkflowDefinitionServiceTest extends
BaseServiceTestTool {
private WorkflowDefinitionLogMapper workflowDefinitionLogMapper;
@Mock
- private WorkflowTaskRelationMapper workflowTaskRelationMapper;
+ private WorkflowTaskRelationDao workflowTaskRelationDao;
@Mock
private WorkflowTaskRelationLogMapper workflowTaskRelationLogMapper;
@@ -289,7 +289,7 @@ public class WorkflowDefinitionServiceTest extends
BaseServiceTestTool {
Mockito.doNothing().when(projectService)
.checkProjectAndAuthThrowException(user, project,
WORKFLOW_BATCH_COPY);
when(workflowDefinitionDao.queryByCodes(definitionCodes)).thenReturn(workflowDefinitionList);
-
when(workflowTaskRelationMapper.queryByWorkflowDefinitionCode(Long.parseLong(codes)))
+
when(workflowTaskRelationDao.queryByWorkflowDefinitionCode(Long.parseLong(codes)))
.thenReturn(workflowTaskRelations);
when(taskDefinitionLogDao.queryTaskDefineLogList(workflowTaskRelations)).thenReturn(taskDefinitionLogs);
when(processService.saveTaskDefine(user, projectCode,
taskDefinitionLogs, true)).thenReturn(1);
@@ -524,7 +524,7 @@ public class WorkflowDefinitionServiceTest extends
BaseServiceTestTool {
}
when(workflowDefinitionDao.queryByCodes(definitionCodes)).thenReturn(workflowDefinitionList);
when(processService.saveWorkflowDefine(user, definition, Boolean.TRUE,
Boolean.TRUE)).thenReturn(2);
-
when(workflowTaskRelationMapper.queryByWorkflowDefinitionCode(processDefinitionCode))
+
when(workflowTaskRelationDao.queryByWorkflowDefinitionCode(processDefinitionCode))
.thenReturn(getProcessTaskRelation());
Assertions.assertDoesNotThrow(() ->
workflowDefinitionService.batchMoveWorkflowDefinition(
@@ -858,7 +858,7 @@ public class WorkflowDefinitionServiceTest extends
BaseServiceTestTool {
when(processService.saveTaskDefine(eq(user), eq(projectCode),
anyList(), eq(Boolean.TRUE))).thenReturn(1);
when(processService.saveWorkflowDefine(any(User.class),
any(WorkflowDefinition.class), eq(Boolean.TRUE),
eq(Boolean.TRUE))).thenReturn(2);
-
when(workflowTaskRelationMapper.queryByWorkflowDefinitionCode(processDefinitionCode))
+
when(workflowTaskRelationDao.queryByWorkflowDefinitionCode(processDefinitionCode))
.thenReturn(Collections.emptyList());
when(processService.saveTaskRelation(eq(user), eq(projectCode),
eq(processDefinitionCode), eq(2), anyList(),
anyList(),
eq(Boolean.TRUE))).thenReturn(Constants.EXIT_CODE_SUCCESS);
diff --git
a/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/repository/WorkflowTaskRelationDao.java
b/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/repository/WorkflowTaskRelationDao.java
new file mode 100644
index 0000000000..d84dd0c796
--- /dev/null
+++
b/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/repository/WorkflowTaskRelationDao.java
@@ -0,0 +1,47 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.dolphinscheduler.dao.repository;
+
+import org.apache.dolphinscheduler.dao.entity.WorkflowTaskRelation;
+
+import java.util.List;
+
+public interface WorkflowTaskRelationDao extends IDao<WorkflowTaskRelation> {
+
+ List<WorkflowTaskRelation> queryByWorkflowDefinitionCode(long
workflowDefinitionCode);
+
+ int batchInsert(List<WorkflowTaskRelation> taskRelationList);
+
+ List<WorkflowTaskRelation> queryUpstreamByCode(long projectCode, long
taskCode);
+
+ List<WorkflowTaskRelation> queryUpstreamByCodes(long projectCode, long
taskCode, Long[] preTaskCodes);
+
+ List<WorkflowTaskRelation>
queryWorkflowTaskRelationsByWorkflowDefinitionCode(long workflowDefinitionCode,
+
Integer workflowDefinitionVersion);
+
+ List<WorkflowTaskRelation> queryDownstreamByWorkflowDefinitionCode(long
workflowDefinitionCode);
+
+ boolean updateWorkflowTaskRelationTaskVersion(WorkflowTaskRelation
workflowTaskRelation);
+
+ void deleteByWorkflowDefinitionCodeAndVersion(long workflowDefinitionCode,
int workflowDefinitionVersion);
+
+ List<WorkflowTaskRelation>
queryWorkflowTaskRelationByTaskCodeAndTaskVersion(long taskCode, long
postTaskVersion);
+
+ List<WorkflowTaskRelation> queryByCode(long projectCode, long
workflowDefinitionCode, long preTaskCode,
+ long postTaskCode);
+}
diff --git
a/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/repository/impl/WorkflowTaskRelationDaoImpl.java
b/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/repository/impl/WorkflowTaskRelationDaoImpl.java
new file mode 100644
index 0000000000..79d84f84c1
--- /dev/null
+++
b/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/repository/impl/WorkflowTaskRelationDaoImpl.java
@@ -0,0 +1,93 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.dolphinscheduler.dao.repository.impl;
+
+import org.apache.dolphinscheduler.dao.entity.WorkflowTaskRelation;
+import org.apache.dolphinscheduler.dao.mapper.WorkflowTaskRelationMapper;
+import org.apache.dolphinscheduler.dao.repository.BaseDao;
+import org.apache.dolphinscheduler.dao.repository.WorkflowTaskRelationDao;
+
+import java.util.List;
+
+import lombok.NonNull;
+
+import org.springframework.stereotype.Repository;
+
+@Repository
+public class WorkflowTaskRelationDaoImpl extends BaseDao<WorkflowTaskRelation,
WorkflowTaskRelationMapper>
+ implements
+ WorkflowTaskRelationDao {
+
+ public WorkflowTaskRelationDaoImpl(@NonNull WorkflowTaskRelationMapper
workflowTaskRelationMapper) {
+ super(workflowTaskRelationMapper);
+ }
+
+ @Override
+ public List<WorkflowTaskRelation> queryByWorkflowDefinitionCode(long
workflowDefinitionCode) {
+ return
mybatisMapper.queryByWorkflowDefinitionCode(workflowDefinitionCode);
+ }
+
+ @Override
+ public int batchInsert(List<WorkflowTaskRelation> taskRelationList) {
+ return mybatisMapper.batchInsert(taskRelationList);
+ }
+
+ @Override
+ public List<WorkflowTaskRelation> queryUpstreamByCode(long projectCode,
long taskCode) {
+ return mybatisMapper.queryUpstreamByCode(projectCode, taskCode);
+ }
+
+ @Override
+ public List<WorkflowTaskRelation> queryUpstreamByCodes(long projectCode,
long taskCode, Long[] preTaskCodes) {
+ return mybatisMapper.queryUpstreamByCodes(projectCode, taskCode,
preTaskCodes);
+ }
+
+ @Override
+ public List<WorkflowTaskRelation>
queryWorkflowTaskRelationsByWorkflowDefinitionCode(long workflowDefinitionCode,
+
Integer workflowDefinitionVersion) {
+ return
mybatisMapper.queryWorkflowTaskRelationsByWorkflowDefinitionCode(workflowDefinitionCode,
+ workflowDefinitionVersion);
+ }
+
+ @Override
+ public List<WorkflowTaskRelation>
queryDownstreamByWorkflowDefinitionCode(long workflowDefinitionCode) {
+ return
mybatisMapper.queryDownstreamByWorkflowDefinitionCode(workflowDefinitionCode);
+ }
+
+ @Override
+ public boolean updateWorkflowTaskRelationTaskVersion(WorkflowTaskRelation
workflowTaskRelation) {
+ return
mybatisMapper.updateWorkflowTaskRelationTaskVersion(workflowTaskRelation) > 0;
+ }
+
+ @Override
+ public void deleteByWorkflowDefinitionCodeAndVersion(long
workflowDefinitionCode, int workflowDefinitionVersion) {
+
mybatisMapper.deleteByWorkflowDefinitionCodeAndVersion(workflowDefinitionCode,
workflowDefinitionVersion);
+ }
+
+ @Override
+ public List<WorkflowTaskRelation>
queryWorkflowTaskRelationByTaskCodeAndTaskVersion(long taskCode,
+
long postTaskVersion) {
+ return
mybatisMapper.queryWorkflowTaskRelationByTaskCodeAndTaskVersion(taskCode,
postTaskVersion);
+ }
+
+ @Override
+ public List<WorkflowTaskRelation> queryByCode(long projectCode, long
workflowDefinitionCode, long preTaskCode,
+ long postTaskCode) {
+ return mybatisMapper.queryByCode(projectCode, workflowDefinitionCode,
preTaskCode, postTaskCode);
+ }
+}
