[DSIP-95][API] Complete the functionality of using dependencies in the complement data (#18003)

Author: det101

dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/executor/workflow/BackfillWorkflowExecutorDelegate.java

@@ -18,6 +18,7 @@
 package org.apache.dolphinscheduler.api.executor.workflow;
 
 import org.apache.dolphinscheduler.api.exceptions.ServiceException;
+import org.apache.dolphinscheduler.api.service.WorkflowLineageService;
 import org.apache.dolphinscheduler.api.validator.workflow.BackfillWorkflowDTO;
 import org.apache.dolphinscheduler.common.enums.ComplementDependentMode;
 import org.apache.dolphinscheduler.common.enums.ExecutionOrder;
@@ -49,17 +50,57 @@
 @Component
 public class BackfillWorkflowExecutorDelegate implements IExecutorDelegate<BackfillWorkflowDTO, List<Integer>> {
 
+    @Autowired
+    private WorkflowLineageService workflowLineageService;
+
     @Autowired
     private RegistryClient registryClient;
 
     @Override
     public List<Integer> execute(final BackfillWorkflowDTO backfillWorkflowDTO) {
+        return executeWithDependentExpansion(backfillWorkflowDTO);
+    }
+
+    /**
+     * Expands optional downstream workflows, then submits root and each downstream in list order.
+     * <p>
+     * {@link RunMode} (serial vs parallel date sharding) is taken only from the <strong>root</strong>
+     * {@code backfillWorkflowDTO}'s {@link BackfillWorkflowDTO.BackfillParamsDTO#getRunMode()}; downstream DTOs
+     * mirror the same mode in their params for consistency.
+     */
+    List<Integer> executeWithDependentExpansion(final BackfillWorkflowDTO backfillWorkflowDTO) {
         // todo: directly call the master api to do backfill
+        List<BackfillWorkflowDTO> dependentBackfillDtos = new ArrayList<>();
+        dependentBackfillDtos.add(backfillWorkflowDTO);
+        if (backfillWorkflowDTO.getBackfillParams()
+                .getBackfillDependentMode() == ComplementDependentMode.ALL_DEPENDENT) {
+
+            List<WorkflowDefinition> downstreamWorkflowList =
+                    workflowLineageService.resolveDownstreamWorkflowDefinitionCodes(
+                            backfillWorkflowDTO.getWorkflowDefinition().getCode(),
+                            backfillWorkflowDTO.getBackfillParams().isAllLevelDependent(),
+                            true);
+            if (downstreamWorkflowList.isEmpty()) {
+                log.info("No downstream dependent workflows found for workflow code {}",
+                        backfillWorkflowDTO.getWorkflowDefinition().getCode());
+            } else {
+                dependentBackfillDtos.addAll(buildResolvedDownstreamBackfillDtos(backfillWorkflowDTO,
+                        backfillWorkflowDTO.getBackfillParams().getBackfillDateList(),
+                        downstreamWorkflowList));
+            }
+        }
+        List<Integer> workflowInstanceIdList = new ArrayList<>();
+        // RunMode is defined by the root request only (not per downstream DTO).
         if (backfillWorkflowDTO.getBackfillParams().getRunMode() == RunMode.RUN_MODE_SERIAL) {
-            return doSerialBackfillWorkflow(backfillWorkflowDTO);
+            for (BackfillWorkflowDTO dependentDto : dependentBackfillDtos) {
+                workflowInstanceIdList.addAll(doSerialBackfillWorkflow(dependentDto));
+            }
         } else {
-            return doParallelBackfillWorkflow(backfillWorkflowDTO);
+            for (BackfillWorkflowDTO dependentDto : dependentBackfillDtos) {
+                workflowInstanceIdList.addAll(doParallelBackfillWorkflow(dependentDto));
+            }
         }
+        return workflowInstanceIdList;
     }
 
     private List<Integer> doSerialBackfillWorkflow(final BackfillWorkflowDTO backfillWorkflowDTO) {
@@ -71,9 +112,7 @@ private List<Integer> doSerialBackfillWorkflow(final BackfillWorkflowDTO backfil
             Collections.sort(backfillTimeList);
         }
 
-        final Integer workflowInstanceId = doBackfillWorkflow(
-                backfillWorkflowDTO,
-                backfillTimeList.stream().map(DateUtils::dateToString).collect(Collectors.toList()));
+        final Integer workflowInstanceId = doBackfillWorkflow(backfillWorkflowDTO, backfillTimeList);
         return Lists.newArrayList(workflowInstanceId);
     }
 
@@ -92,8 +131,7 @@ private List<Integer> doParallelBackfillWorkflow(final BackfillWorkflowDTO backf
         final List<Integer> workflowInstanceIdList = Lists.newArrayList();
         for (List<ZonedDateTime> stringDate : splitDateTime(listDate, expectedParallelismNumber)) {
             final Integer workflowInstanceId = doBackfillWorkflow(
-                    backfillWorkflowDTO,
-                    stringDate.stream().map(DateUtils::dateToString).collect(Collectors.toList()));
+                    backfillWorkflowDTO, stringDate);
             workflowInstanceIdList.add(workflowInstanceId);
         }
         return workflowInstanceIdList;
@@ -124,12 +162,15 @@ private List<List<ZonedDateTime>> splitDateTime(List<ZonedDateTime> dateTimeList
     }
 
     private Integer doBackfillWorkflow(final BackfillWorkflowDTO backfillWorkflowDTO,
-                                       final List<String> backfillTimeList) {
+                                       final List<ZonedDateTime> backfillDateTimes) {
         final Server masterServer = registryClient.getRandomServer(RegistryNodeType.MASTER).orElse(null);
         if (masterServer == null) {
             throw new ServiceException("no master server available");
         }
 
+        final List<String> backfillTimeList =
+                backfillDateTimes.stream().map(DateUtils::dateToString).collect(Collectors.toList());
+
         final WorkflowDefinition workflowDefinition = backfillWorkflowDTO.getWorkflowDefinition();
         final WorkflowBackfillTriggerRequest backfillTriggerRequest = WorkflowBackfillTriggerRequest.builder()
                 .userId(backfillWorkflowDTO.getLoginUser().getId())
@@ -149,22 +190,76 @@ private Integer doBackfillWorkflow(final BackfillWorkflowDTO backfillWorkflowDTO
                 .dryRun(backfillWorkflowDTO.getDryRun())
                 .build();
 
-        final WorkflowBackfillTriggerResponse backfillTriggerResponse = Clients
-                .withService(IWorkflowControlClient.class)
-                .withHost(masterServer.getHost() + ":" + masterServer.getPort())
-                .backfillTriggerWorkflow(backfillTriggerRequest);
+        final WorkflowBackfillTriggerResponse backfillTriggerResponse =
+                triggerBackfillWorkflow(backfillTriggerRequest, masterServer);
         if (!backfillTriggerResponse.isSuccess()) {
             throw new ServiceException("Backfill workflow failed: " + backfillTriggerResponse.getMessage());
         }
-        final BackfillWorkflowDTO.BackfillParamsDTO backfillParams = backfillWorkflowDTO.getBackfillParams();
-        if (backfillParams.getBackfillDependentMode() == ComplementDependentMode.ALL_DEPENDENT) {
-            doBackfillDependentWorkflow(backfillWorkflowDTO, backfillTimeList);
-        }
         return backfillTriggerResponse.getWorkflowInstanceId();
     }
 
-    private void doBackfillDependentWorkflow(final BackfillWorkflowDTO backfillWorkflowDTO,
-                                             final List<String> backfillTimeList) {
-        // todo:
+    protected WorkflowBackfillTriggerResponse triggerBackfillWorkflow(final WorkflowBackfillTriggerRequest request,
+                                                                      final Server masterServer) {
+        return Clients
+                .withService(IWorkflowControlClient.class)
+                .withHost(masterServer.getHost() + ":" + masterServer.getPort())
+                .backfillTriggerWorkflow(request);
+    }
+
+    /**
+     * Builds {@link BackfillWorkflowDTO} list for resolved downstream workflows.
+     * {@link RunMode} in each downstream {@link BackfillWorkflowDTO.BackfillParamsDTO} matches the root (see
+     * {@link #executeWithDependentExpansion(BackfillWorkflowDTO)}).
+     */
+    private List<BackfillWorkflowDTO> buildResolvedDownstreamBackfillDtos(final BackfillWorkflowDTO backfillWorkflowDTO,
+                                                                          final List<ZonedDateTime> backfillDateTimes,
+                                                                          final List<WorkflowDefinition> downstreamWorkflows) {
+        final long upstreamWorkflowCode = backfillWorkflowDTO.getWorkflowDefinition().getCode();
+        final List<ZonedDateTime> upstreamBackfillDates = new ArrayList<>(backfillDateTimes);
+        final BackfillWorkflowDTO.BackfillParamsDTO originalParams = backfillWorkflowDTO.getBackfillParams();
+        final boolean allLevelDependent = originalParams.isAllLevelDependent();
+
+        final List<BackfillWorkflowDTO> result = new ArrayList<>();
+        for (WorkflowDefinition downstreamWorkflow : downstreamWorkflows) {
+            final long downstreamCode = downstreamWorkflow.getCode();
+
+            final BackfillWorkflowDTO.BackfillParamsDTO dependentParams =
+                    BackfillWorkflowDTO.BackfillParamsDTO.builder()
+                            // Same as root; executor also branches on root RunMode only.
+                            .runMode(originalParams.getRunMode())
+                            .backfillDateList(upstreamBackfillDates)
+                            .expectedParallelismNumber(originalParams.getExpectedParallelismNumber())
+                            // Downstream expansion has already been decided in resolution stage.
+                            .backfillDependentMode(ComplementDependentMode.OFF_MODE)
+                            .allLevelDependent(allLevelDependent)
+                            .executionOrder(originalParams.getExecutionOrder())
+                            .build();
+
+            final BackfillWorkflowDTO dependentBackfillDTO = BackfillWorkflowDTO.builder()
+                    .loginUser(backfillWorkflowDTO.getLoginUser())
+                    .workflowDefinition(downstreamWorkflow)
+                    .startNodes(null)
+                    .failureStrategy(backfillWorkflowDTO.getFailureStrategy())
+                    .taskDependType(backfillWorkflowDTO.getTaskDependType())
+                    .execType(backfillWorkflowDTO.getExecType())
+                    .warningType(backfillWorkflowDTO.getWarningType())
+                    .warningGroupId(downstreamWorkflow.getWarningGroupId())
+                    .runMode(dependentParams.getRunMode())
+                    .workflowInstancePriority(backfillWorkflowDTO.getWorkflowInstancePriority())
+                    .workerGroup(backfillWorkflowDTO.getWorkerGroup())
+                    .tenantCode(backfillWorkflowDTO.getTenantCode())
+                    .environmentCode(backfillWorkflowDTO.getEnvironmentCode())
+                    .startParamList(backfillWorkflowDTO.getStartParamList())
+                    .dryRun(backfillWorkflowDTO.getDryRun())
+                    .backfillParams(dependentParams)
+                    .build();
+
+            log.info("Built dependent backfill DTO for workflow {} (upstream {}) with backfill dates {}",
+                    downstreamCode, upstreamWorkflowCode,
+                    backfillDateTimes.stream().map(DateUtils::dateToString).collect(Collectors.toList()));
+
+            result.add(dependentBackfillDTO);
+        }
+        return result;
     }
 }

dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/WorkflowLineageService.java

@@ -21,6 +21,7 @@
 import org.apache.dolphinscheduler.dao.entity.DependentWorkflowDefinition;
 import org.apache.dolphinscheduler.dao.entity.WorkFlowLineage;
 import org.apache.dolphinscheduler.dao.entity.WorkFlowRelationDetail;
+import org.apache.dolphinscheduler.dao.entity.WorkflowDefinition;
 import org.apache.dolphinscheduler.dao.entity.WorkflowTaskLineage;
 
 import java.util.List;
@@ -42,6 +43,37 @@ public interface WorkflowLineageService {
      */
     List<DependentWorkflowDefinition> queryDownstreamDependentWorkflowDefinitions(Long workflowDefinitionCode);
 
+    /**
+     * Resolve downstream workflow definitions that depend on the given root workflow, using stored lineage.
+     * <ul>
+     *     <li>{@code allLevelDependent == false}: only direct dependents of the root.</li>
+     *     <li>{@code allLevelDependent == true}: transitive dependents (BFS over direct dependents), skipping the root
+     *     when it reappears as an edge target (cycle back to root).</li>
+     * </ul>
+     *
+     * @param rootWorkflowDefinitionCode workflow to start from (not included in the result)
+     * @param allLevelDependent          {@code true} for transitive closure, {@code false} for one hop only
+     * @return ordered distinct downstream workflow definitions (stable order: BFS / insertion order)
+     */
+    default List<WorkflowDefinition> resolveDownstreamWorkflowDefinitionCodes(long rootWorkflowDefinitionCode,
+                                                                              boolean allLevelDependent) {
+        return resolveDownstreamWorkflowDefinitionCodes(rootWorkflowDefinitionCode, allLevelDependent, false);
+    }
+
+    /**
+     * Resolve downstream workflow definitions and optionally filter offline workflows.
+     * When {@code filterOfflineWorkflow} is true, offline workflow definitions are excluded from the result and are
+     * not expanded further during transitive traversal.
+     *
+     * @param rootWorkflowDefinitionCode workflow to start from (not included in the result)
+     * @param allLevelDependent          {@code true} for transitive closure, {@code false} for one hop only
+     * @param filterOfflineWorkflow      whether offline workflows should be filtered out during traversal
+     * @return ordered distinct downstream workflow definitions (stable order: BFS / insertion order)
+     */
+    List<WorkflowDefinition> resolveDownstreamWorkflowDefinitionCodes(long rootWorkflowDefinitionCode,
+                                                                      boolean allLevelDependent,
+                                                                      boolean filterOfflineWorkflow);
+
     /**
      * Query and return tasks dependence with string format, is a wrapper of queryTaskDepOnTask and task query method.
      *

dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/WorkflowLineageServiceImpl.java

@@ -21,6 +21,7 @@
 import org.apache.dolphinscheduler.api.exceptions.ServiceException;
 import org.apache.dolphinscheduler.api.service.WorkflowLineageService;
 import org.apache.dolphinscheduler.common.constants.Constants;
+import org.apache.dolphinscheduler.common.enums.ReleaseState;
 import org.apache.dolphinscheduler.dao.entity.DependentLineageTask;
 import org.apache.dolphinscheduler.dao.entity.DependentWorkflowDefinition;
 import org.apache.dolphinscheduler.dao.entity.Project;
@@ -37,9 +38,15 @@
 
 import java.text.MessageFormat;
 import java.util.ArrayList;
+import java.util.Collection;
 import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.LinkedHashSet;
 import java.util.List;
+import java.util.Map;
 import java.util.Optional;
+import java.util.Set;
 import java.util.stream.Collectors;
 import java.util.stream.Stream;
 
@@ -319,6 +326,112 @@ public List<DependentLineageTask> queryDependentWorkflowDefinitions(long project
         return dependentLineageTaskList;
     }
 
+    @Override
+    public List<WorkflowDefinition> resolveDownstreamWorkflowDefinitionCodes(long rootWorkflowDefinitionCode,
+                                                                             boolean allLevelDependent,
+                                                                             boolean filterOfflineWorkflow) {
+
+        Set<Long> resultCodes = new LinkedHashSet<>();
+        Set<Long> visitedWorkflowCodes = new HashSet<>();
+        List<Long> frontier = Collections.singletonList(rootWorkflowDefinitionCode);
+        Map<Long, WorkflowDefinition> workflowDefinitionCache = new HashMap<>();
+
+        while (!CollectionUtils.isEmpty(frontier)) {
+            List<Long> currentLevel = new ArrayList<>(frontier);
+            frontier = new ArrayList<>();
+
+            for (Long upstreamCode : currentLevel) {
+                if (!visitedWorkflowCodes.add(upstreamCode)) {
+                    continue;
+                }
+                List<Long> directDownstreamCodes = queryDirectDownstreamWorkflowCodes(upstreamCode);
+                if (CollectionUtils.isEmpty(directDownstreamCodes)) {
+                    continue;
+                }
+
+                cacheWorkflowDefinitions(directDownstreamCodes, workflowDefinitionCache);
+                for (Long downstreamCode : directDownstreamCodes) {
+                    if (downstreamCode == rootWorkflowDefinitionCode) {
+                        // Skip cycle edge back to root; root should never appear in downstream result.
+                        continue;
+                    }
+                    WorkflowDefinition downstreamWorkflow = workflowDefinitionCache.get(downstreamCode);
+                    if (downstreamWorkflow == null) {
+                        continue;
+                    }
+                    if (!resultCodes.add(downstreamCode)) {
+                        continue;
+                    }
+                    if (allLevelDependent && (!filterOfflineWorkflow
+                            || downstreamWorkflow.getReleaseState() != ReleaseState.OFFLINE)) {
+                        frontier.add(downstreamCode);
+                    }
+                }
+            }
+
+            if (!allLevelDependent) {
+                break;
+            }
+        }
+
+        if (resultCodes.isEmpty()) {
+            return Collections.emptyList();
+        }
+
+        if (filterOfflineWorkflow) {
+            resultCodes = resultCodes.stream()
+                    .filter(code -> {
+                        WorkflowDefinition definition = workflowDefinitionCache.get(code);
+                        return definition != null
+                                && definition.getReleaseState() != ReleaseState.OFFLINE;
+                    })
+                    .collect(Collectors.toCollection(LinkedHashSet::new));
+            if (resultCodes.isEmpty()) {
+                return Collections.emptyList();
+            }
+        }
+
+        List<WorkflowDefinition> orderedDefinitions = new ArrayList<>();
+        for (Long code : resultCodes) {
+            WorkflowDefinition definition = workflowDefinitionCache.get(code);
+            if (definition != null) {
+                orderedDefinitions.add(definition);
+            }
+        }
+        return orderedDefinitions;
+    }
+
+    private List<Long> queryDirectDownstreamWorkflowCodes(long upstreamWorkflowDefinitionCode) {
+        List<WorkflowTaskLineage> workflowTaskLineageList =
+                workflowTaskLineageDao.queryWorkFlowLineageByDept(Constants.DEFAULT_PROJECT_CODE,
+                        upstreamWorkflowDefinitionCode,
+                        Constants.DEPENDENT_ALL_TASK);
+        if (CollectionUtils.isEmpty(workflowTaskLineageList)) {
+            return Collections.emptyList();
+        }
+        return workflowTaskLineageList.stream()
+                .map(WorkflowTaskLineage::getWorkflowDefinitionCode)
+                .distinct()
+                .collect(Collectors.toList());
+    }
+
+    private void cacheWorkflowDefinitions(Collection<Long> workflowDefinitionCodes,
+                                          Map<Long, WorkflowDefinition> workflowDefinitionCache) {
+        List<Long> missingCodes = workflowDefinitionCodes.stream()
+                .filter(code -> !workflowDefinitionCache.containsKey(code))
+                .collect(Collectors.toList());
+        if (CollectionUtils.isEmpty(missingCodes)) {
+            return;
+        }
+        List<WorkflowDefinition> workflowDefinitions = workflowDefinitionMapper.queryByCodes(missingCodes);
+        if (CollectionUtils.isEmpty(workflowDefinitions)) {
+            return;
+        }
+        for (WorkflowDefinition workflowDefinition : workflowDefinitions) {
+            workflowDefinitionCache.put(workflowDefinition.getCode(), workflowDefinition);
+        }
+    }
+
     @Override
     public int updateWorkflowLineage(long workflowDefinitionCode, List<WorkflowTaskLineage> workflowTaskLineages) {
         // Remove existing lineage first to keep data consistent