fix(api): isolate visited codes per parallel backfill chunk

Ensure dependent backfill triggering is evaluated per parallel date chunk, and batch-load downstream workflow definitions to avoid N+1 queries. Add regression coverage for parallel mode visited-code isolation.

Made-with: Cursor

Author: luxl

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

@@ -42,7 +42,9 @@
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.HashSet;
+import java.util.LinkedHashSet;
 import java.util.List;
+import java.util.Map;
 import java.util.Set;
 import java.util.stream.Collectors;
 
@@ -118,11 +120,14 @@ private List<Integer> doParallelBackfillWorkflow(final BackfillWorkflowDTO backf
 
         log.info("In parallel mode, current expectedParallelismNumber: {}", expectedParallelismNumber);
         final List<Integer> workflowInstanceIdList = Lists.newArrayList();
+        final Set<Long> baseVisitedCodes = visitedCodes == null ? new HashSet<>() : visitedCodes;
         for (List<ZonedDateTime> stringDate : splitDateTime(listDate, expectedParallelismNumber)) {
+            // Each parallel chunk should keep its own traversal context to avoid cross-chunk pollution.
+            final Set<Long> chunkVisitedCodes = new HashSet<>(baseVisitedCodes);
             final Integer workflowInstanceId = doBackfillWorkflow(
                     backfillWorkflowDTO,
                     stringDate.stream().map(DateUtils::dateToString).collect(Collectors.toList()),
-                    visitedCodes);
+                    chunkVisitedCodes);
             workflowInstanceIdList.add(workflowInstanceId);
         }
         return workflowInstanceIdList;
@@ -190,11 +195,17 @@ private Integer doBackfillWorkflow(final BackfillWorkflowDTO backfillWorkflowDTO
         if (backfillParams.getBackfillDependentMode() == ComplementDependentMode.ALL_DEPENDENT) {
             final Set<Long> effectiveVisitedCodes = visitedCodes == null ? new HashSet<>() : visitedCodes;
             effectiveVisitedCodes.add(backfillWorkflowDTO.getWorkflowDefinition().getCode());
-            doBackfillDependentWorkflow(backfillWorkflowDTO, backfillTimeList, effectiveVisitedCodes);
+            doBackfillDependentWorkflowForTesting(backfillWorkflowDTO, backfillTimeList, effectiveVisitedCodes);
         }
         return backfillTriggerResponse.getWorkflowInstanceId();
     }
 
+    void doBackfillDependentWorkflowForTesting(final BackfillWorkflowDTO backfillWorkflowDTO,
+                                               final List<String> backfillTimeList,
+                                               final Set<Long> visitedCodes) {
+        doBackfillDependentWorkflow(backfillWorkflowDTO, backfillTimeList, visitedCodes);
+    }
+
     private void doBackfillDependentWorkflow(final BackfillWorkflowDTO backfillWorkflowDTO,
                                              final List<String> backfillTimeList,
                                              final Set<Long> visitedCodes) {
@@ -209,6 +220,12 @@ private void doBackfillDependentWorkflow(final BackfillWorkflowDTO backfillWorkf
             log.info("No downstream dependent workflows found for workflow code {}", upstreamWorkflowCode);
             return;
         }
+        final Set<Long> downstreamCodes = downstreamDefinitions.stream()
+                .map(DependentWorkflowDefinition::getWorkflowDefinitionCode)
+                .collect(Collectors.toCollection(LinkedHashSet::new));
+        final List<WorkflowDefinition> downstreamWorkflowList = workflowDefinitionDao.queryByCodes(downstreamCodes);
+        final Map<Long, WorkflowDefinition> downstreamWorkflowMap = downstreamWorkflowList.stream()
+                .collect(Collectors.toMap(WorkflowDefinition::getCode, workflow -> workflow));
 
         // 2) Convert upstream backfill time from string to ZonedDateTime as the base business dates for downstream
         // backfill
@@ -226,8 +243,7 @@ private void doBackfillDependentWorkflow(final BackfillWorkflowDTO backfillWorkf
                 continue;
             }
 
-            WorkflowDefinition downstreamWorkflow =
-                    workflowDefinitionDao.queryByCode(downstreamCode).orElse(null);
+            WorkflowDefinition downstreamWorkflow = downstreamWorkflowMap.get(downstreamCode);
             if (downstreamWorkflow == null) {
                 log.warn("Skip dependent workflow {}, workflow definition not found", downstreamCode);
                 continue;

dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/executor/workflow/BackfillWorkflowExecutorDelegateTest.java

@@ -18,7 +18,7 @@
 package org.apache.dolphinscheduler.api.executor.workflow;
 
 import static org.mockito.ArgumentMatchers.any;
-import static org.mockito.ArgumentMatchers.anyLong;
+import static org.mockito.Mockito.doAnswer;
 import static org.mockito.Mockito.doReturn;
 import static org.mockito.Mockito.never;
 import static org.mockito.Mockito.times;
@@ -40,8 +40,8 @@
 import java.util.Arrays;
 import java.util.Collections;
 import java.util.HashSet;
+import java.util.LinkedHashSet;
 import java.util.List;
-import java.util.Optional;
 import java.util.Set;
 import java.util.stream.Collectors;
 
@@ -102,7 +102,7 @@ public void testDoBackfillDependentWorkflow_NoDownstreamDefinitions() throws Exc
         visitedCodes.add(dto.getWorkflowDefinition().getCode());
         method.invoke(backfillWorkflowExecutorDelegate, dto, backfillTimeList, visitedCodes);
 
-        verify(workflowDefinitionDao, never()).queryByCode(anyLong());
+        verify(workflowDefinitionDao, never()).queryByCodes(any());
     }
 
     @Test
@@ -145,8 +145,8 @@ public void testDoBackfillDependentWorkflow_WithDownstream_AllLevelDependent() t
 
         when(workflowLineageService.queryDownstreamDependentWorkflowDefinitions(upstreamCode))
                 .thenReturn(Arrays.asList(selfDependent, validDependent));
-        when(workflowDefinitionDao.queryByCode(downstreamCode))
-                .thenReturn(Optional.of(downstreamWorkflow));
+        when(workflowDefinitionDao.queryByCodes(new LinkedHashSet<>(Arrays.asList(upstreamCode, downstreamCode))))
+                .thenReturn(Collections.singletonList(downstreamWorkflow));
 
         ArgumentCaptor<BackfillWorkflowDTO> captor = ArgumentCaptor.forClass(BackfillWorkflowDTO.class);
         doReturn(Collections.singletonList(1)).when(backfillWorkflowExecutorDelegate)
@@ -164,7 +164,8 @@ public void testDoBackfillDependentWorkflow_WithDownstream_AllLevelDependent() t
         visitedCodes.add(dto.getWorkflowDefinition().getCode());
         method.invoke(backfillWorkflowExecutorDelegate, dto, backfillTimeList, visitedCodes);
 
-        verify(workflowDefinitionDao).queryByCode(downstreamCode);
+        verify(workflowDefinitionDao)
+                .queryByCodes(new LinkedHashSet<>(Arrays.asList(upstreamCode, downstreamCode)));
 
         BackfillWorkflowDTO captured = captor.getValue();
         Assertions.assertNotNull(captured);
@@ -219,8 +220,8 @@ public void testDoBackfillDependentWorkflow_WithDownstream_SingleLevelDependent(
 
         when(workflowLineageService.queryDownstreamDependentWorkflowDefinitions(upstreamCode))
                 .thenReturn(Collections.singletonList(validDependent));
-        when(workflowDefinitionDao.queryByCode(downstreamCode))
-                .thenReturn(Optional.of(downstreamWorkflow));
+        when(workflowDefinitionDao.queryByCodes(Collections.singleton(downstreamCode)))
+                .thenReturn(Collections.singletonList(downstreamWorkflow));
 
         ArgumentCaptor<BackfillWorkflowDTO> captor = ArgumentCaptor.forClass(BackfillWorkflowDTO.class);
         doReturn(Collections.singletonList(1)).when(backfillWorkflowExecutorDelegate)
@@ -236,7 +237,7 @@ public void testDoBackfillDependentWorkflow_WithDownstream_SingleLevelDependent(
         visitedCodes.add(dto.getWorkflowDefinition().getCode());
         method.invoke(backfillWorkflowExecutorDelegate, dto, backfillTimeList, visitedCodes);
 
-        verify(workflowDefinitionDao).queryByCode(downstreamCode);
+        verify(workflowDefinitionDao).queryByCodes(Collections.singleton(downstreamCode));
 
         BackfillWorkflowDTO captured = captor.getValue();
         Assertions.assertNotNull(captured);
@@ -275,7 +276,8 @@ public void testDoBackfillDependentWorkflow_SkipWorkflowNotFound() throws Except
 
         when(workflowLineageService.queryDownstreamDependentWorkflowDefinitions(upstreamCode))
                 .thenReturn(Collections.singletonList(dep));
-        when(workflowDefinitionDao.queryByCode(downstreamCode)).thenReturn(Optional.empty());
+        when(workflowDefinitionDao.queryByCodes(Collections.singleton(downstreamCode)))
+                .thenReturn(Collections.emptyList());
 
         Method method = BackfillWorkflowExecutorDelegate.class.getDeclaredMethod(
                 "doBackfillDependentWorkflow", BackfillWorkflowDTO.class, List.class, Set.class);
@@ -318,7 +320,8 @@ public void testDoBackfillDependentWorkflow_SkipOfflineWorkflow() throws Excepti
 
         when(workflowLineageService.queryDownstreamDependentWorkflowDefinitions(upstreamCode))
                 .thenReturn(Collections.singletonList(dep));
-        when(workflowDefinitionDao.queryByCode(downstreamCode)).thenReturn(Optional.of(offlineDownstream));
+        when(workflowDefinitionDao.queryByCodes(Collections.singleton(downstreamCode)))
+                .thenReturn(Collections.singletonList(offlineDownstream));
 
         Method method = BackfillWorkflowExecutorDelegate.class.getDeclaredMethod(
                 "doBackfillDependentWorkflow", BackfillWorkflowDTO.class, List.class, Set.class);
@@ -371,8 +374,10 @@ public void testDoBackfillDependentWorkflow_MultiLevelAndCycle() throws Exceptio
                 .thenReturn(Collections.singletonList(depToB));
         when(workflowLineageService.queryDownstreamDependentWorkflowDefinitions(workflowB))
                 .thenReturn(Arrays.asList(depToA, depToC));
-        when(workflowDefinitionDao.queryByCode(workflowB)).thenReturn(Optional.of(downstreamB));
-        when(workflowDefinitionDao.queryByCode(workflowC)).thenReturn(Optional.of(downstreamC));
+        when(workflowDefinitionDao.queryByCodes(Collections.singleton(workflowB)))
+                .thenReturn(Collections.singletonList(downstreamB));
+        when(workflowDefinitionDao.queryByCodes(new LinkedHashSet<>(Arrays.asList(workflowA, workflowC))))
+                .thenReturn(Collections.singletonList(downstreamC));
 
         ArgumentCaptor<BackfillWorkflowDTO> captor = ArgumentCaptor.forClass(BackfillWorkflowDTO.class);
         doReturn(Collections.singletonList(1)).when(backfillWorkflowExecutorDelegate)
@@ -394,7 +399,7 @@ public void testDoBackfillDependentWorkflow_MultiLevelAndCycle() throws Exceptio
         method.invoke(backfillWorkflowExecutorDelegate, dtoB, backfillTimeList, visitedCodes);
 
         verify(backfillWorkflowExecutorDelegate, times(2)).executeWithVisitedCodes(any(), any());
-        verify(workflowDefinitionDao, never()).queryByCode(workflowA);
+        verify(workflowDefinitionDao, times(2)).queryByCodes(any());
 
         List<Long> triggeredCodes = captor.getAllValues().stream()
                 .map(it -> it.getWorkflowDefinition().getCode())
@@ -403,4 +408,46 @@ public void testDoBackfillDependentWorkflow_MultiLevelAndCycle() throws Exceptio
         Assertions.assertTrue(visitedCodes.contains(workflowB));
         Assertions.assertTrue(visitedCodes.contains(workflowC));
     }
+
+    @Test
+    public void testDoParallelBackfillWorkflow_ShouldIsolateVisitedCodesAcrossChunks() {
+        long upstreamCode = 500L;
+        WorkflowDefinition upstreamWorkflow =
+                WorkflowDefinition.builder().code(upstreamCode).releaseState(ReleaseState.ONLINE).build();
+        List<ZonedDateTime> dates = Arrays.asList(
+                ZonedDateTime.parse("2026-02-01T00:00:00Z"),
+                ZonedDateTime.parse("2026-02-02T00:00:00Z"),
+                ZonedDateTime.parse("2026-02-03T00:00:00Z"));
+        BackfillWorkflowDTO.BackfillParamsDTO params = BackfillWorkflowDTO.BackfillParamsDTO.builder()
+                .runMode(RunMode.RUN_MODE_PARALLEL)
+                .backfillDateList(dates)
+                .expectedParallelismNumber(2)
+                .backfillDependentMode(ComplementDependentMode.ALL_DEPENDENT)
+                .allLevelDependent(true)
+                .executionOrder(ExecutionOrder.ASC_ORDER)
+                .build();
+        BackfillWorkflowDTO dto = BackfillWorkflowDTO.builder()
+                .workflowDefinition(upstreamWorkflow)
+                .backfillParams(params)
+                .build();
+        Set<Long> baseVisitedCodes = new HashSet<>(Collections.singleton(upstreamCode));
+        List<Set<Long>> visitedSnapshotPerChunk = new java.util.ArrayList<>();
+
+        doAnswer(invocation -> {
+            Set<Long> chunkVisited = invocation.getArgument(2);
+            visitedSnapshotPerChunk.add(new HashSet<>(chunkVisited));
+            chunkVisited.add(9000L + visitedSnapshotPerChunk.size());
+            return null;
+        }).when(backfillWorkflowExecutorDelegate).doBackfillDependentWorkflowForTesting(any(), any(), any());
+
+        List<Integer> result = backfillWorkflowExecutorDelegate.executeWithVisitedCodes(dto, baseVisitedCodes);
+
+        Assertions.assertEquals(2, result.size());
+        Assertions.assertEquals(2, visitedSnapshotPerChunk.size());
+        Assertions.assertEquals(Collections.singleton(upstreamCode), visitedSnapshotPerChunk.get(0));
+        Assertions.assertEquals(Collections.singleton(upstreamCode), visitedSnapshotPerChunk.get(1));
+        Assertions.assertEquals(Collections.singleton(upstreamCode), baseVisitedCodes);
+        verify(backfillWorkflowExecutorDelegate, times(2))
+                .doBackfillDependentWorkflowForTesting(any(), any(), any());
+    }
 }