[Fix-16879] Remove parent VarPool from sub-workflow start params to prevent parameter duplication

dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/executor/plugin/subworkflow/SubWorkflowLogicTask.java

@@ -237,7 +237,7 @@ private SubWorkflowLogicTaskRuntimeContext triggerNewSubWorkflow() {
         final List<Property> paramList = mergeParams(asList(
                 new ArrayList<>(deserializeVarPool(workflowInstance.getGlobalParams())),
                 commandParam.getCommandParams(),
-                new ArrayList<>(deserializeVarPool(workflowInstance.getVarPool()))));
+                taskExecutionContext.getVarPool()));
 
         final WorkflowManualTriggerRequest workflowManualTriggerRequest = WorkflowManualTriggerRequest.builder()
                 .userId(taskExecutionContext.getExecutorId())
@@ -261,7 +261,7 @@ private SubWorkflowLogicTaskRuntimeContext triggerNewSubWorkflow() {
         return SubWorkflowLogicTaskRuntimeContext.of(subWorkflowInstanceId);
     }
 
-    private List<Property> mergeParams(List<List<Property>> params) {
+    static List<Property> mergeParams(List<List<Property>> params) {
         if (CollectionUtils.isEmpty(params)) {
             return Collections.emptyList();
         }

dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/task/execution/TaskExecutionContextBuilder.java

@@ -29,6 +29,7 @@
 import org.apache.dolphinscheduler.plugin.task.api.enums.TaskTimeoutStrategy;
 import org.apache.dolphinscheduler.plugin.task.api.model.Property;
 import org.apache.dolphinscheduler.plugin.task.api.parameters.resource.ResourceParametersHelper;
+import org.apache.dolphinscheduler.plugin.task.api.utils.VarPoolUtils;
 
 import java.util.Map;
 import java.util.concurrent.TimeUnit;
@@ -67,6 +68,7 @@ public TaskExecutionContextBuilder buildTaskInstanceRelatedInfo(final TaskInstan
         taskExecutionContext.setCpuQuota(taskInstance.getCpuQuota());
         taskExecutionContext.setMemoryMax(taskInstance.getMemoryMax());
         taskExecutionContext.setAppIds(taskInstance.getAppLink());
+        taskExecutionContext.setVarPool(VarPoolUtils.deserializeVarPool(taskInstance.getVarPool()));
         return this;
     }
 

dolphinscheduler-master/src/test/java/org/apache/dolphinscheduler/server/master/engine/executor/plugin/subworkflow/SubWorkflowLogicTaskMergeParamsTest.java

@@ -0,0 +1,193 @@
+/*
+ * 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.server.master.engine.executor.plugin.subworkflow;
+
+import static com.google.common.truth.Truth.assertThat;
+
+import org.apache.dolphinscheduler.plugin.task.api.enums.DataType;
+import org.apache.dolphinscheduler.plugin.task.api.enums.Direct;
+import org.apache.dolphinscheduler.plugin.task.api.model.Property;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+
+import org.junit.jupiter.api.DisplayName;
+import org.junit.jupiter.api.Test;
+
+class SubWorkflowLogicTaskMergeParamsTest {
+
+    @Test
+    @DisplayName("Test mergeParams: upstream VarPool (OUT) overrides global parameter (IN)")
+    void testMergeParams_upstreamVarPoolOverridesGlobalParam() {
+        // Simulates: globalParams=[param1=global_val], varPool=[param1=upstream_val]
+        // The varPool (last in merge order) should win
+        final Property globalParam = Property.builder()
+                .prop("param1")
+                .direct(Direct.IN)
+                .type(DataType.VARCHAR)
+                .value("global_val")
+                .build();
+        final Property upstreamOutParam = Property.builder()
+                .prop("param1")
+                .direct(Direct.OUT)
+                .type(DataType.VARCHAR)
+                .value("upstream_val")
+                .build();
+
+        final List<Property> result = SubWorkflowLogicTask.mergeParams(Arrays.asList(
+                Collections.singletonList(globalParam),
+                Collections.emptyList(),
+                Collections.singletonList(upstreamOutParam)));
+
+        assertThat(result).hasSize(1);
+        assertThat(result.get(0).getProp()).isEqualTo("param1");
+        assertThat(result.get(0).getValue()).isEqualTo("upstream_val");
+    }
+
+    @Test
+    @DisplayName("Test mergeParams: command (start) parameter overrides global parameter")
+    void testMergeParams_commandParamOverridesGlobalParam() {
+        // Simulates: globalParams=[param1=global_val], commandParams=[param1=start_val]
+        final Property globalParam = Property.builder()
+                .prop("param1")
+                .direct(Direct.IN)
+                .type(DataType.VARCHAR)
+                .value("global_val")
+                .build();
+        final Property commandParam = Property.builder()
+                .prop("param1")
+                .direct(Direct.IN)
+                .type(DataType.VARCHAR)
+                .value("start_val")
+                .build();
+
+        final List<Property> result = SubWorkflowLogicTask.mergeParams(Arrays.asList(
+                Collections.singletonList(globalParam),
+                Collections.singletonList(commandParam),
+                Collections.emptyList()));
+
+        assertThat(result).hasSize(1);
+        assertThat(result.get(0).getProp()).isEqualTo("param1");
+        assertThat(result.get(0).getValue()).isEqualTo("start_val");
+    }
+
+    @Test
+    @DisplayName("Test mergeParams: upstream VarPool overrides command (start) parameter")
+    void testMergeParams_varPoolOverridesCommandParam() {
+        // Simulates: commandParams=[param1=start_val], varPool=[param1=upstream_val]
+        // The varPool (last in merge order) should win
+        final Property commandParam = Property.builder()
+                .prop("param1")
+                .direct(Direct.IN)
+                .type(DataType.VARCHAR)
+                .value("start_val")
+                .build();
+        final Property upstreamOutParam = Property.builder()
+                .prop("param1")
+                .direct(Direct.OUT)
+                .type(DataType.VARCHAR)
+                .value("upstream_val")
+                .build();
+
+        final List<Property> result = SubWorkflowLogicTask.mergeParams(Arrays.asList(
+                Collections.emptyList(),
+                Collections.singletonList(commandParam),
+                Collections.singletonList(upstreamOutParam)));
+
+        assertThat(result).hasSize(1);
+        assertThat(result.get(0).getProp()).isEqualTo("param1");
+        assertThat(result.get(0).getValue()).isEqualTo("upstream_val");
+    }
+
+    @Test
+    @DisplayName("Test mergeParams: full precedence — global < command < upstream VarPool")
+    void testMergeParams_fullPrecedence() {
+        // Simulates a conflict where all three sources provide the same key
+        // globalParams=[param1=global_val], commandParams=[param1=start_val], varPool=[param1=upstream_val]
+        // The varPool (last in merge order) should win
+        final Property globalParam = Property.builder()
+                .prop("param1")
+                .direct(Direct.IN)
+                .type(DataType.VARCHAR)
+                .value("global_val")
+                .build();
+        final Property commandParam = Property.builder()
+                .prop("param1")
+                .direct(Direct.IN)
+                .type(DataType.VARCHAR)
+                .value("start_val")
+                .build();
+        final Property upstreamOutParam = Property.builder()
+                .prop("param1")
+                .direct(Direct.OUT)
+                .type(DataType.VARCHAR)
+                .value("upstream_val")
+                .build();
+
+        final List<Property> result = SubWorkflowLogicTask.mergeParams(Arrays.asList(
+                Collections.singletonList(globalParam),
+                Collections.singletonList(commandParam),
+                Collections.singletonList(upstreamOutParam)));
+
+        assertThat(result).hasSize(1);
+        assertThat(result.get(0).getProp()).isEqualTo("param1");
+        assertThat(result.get(0).getValue()).isEqualTo("upstream_val");
+    }
+
+    @Test
+    @DisplayName("Test mergeParams: non-conflicting parameters from all sources are preserved")
+    void testMergeParams_nonConflictingParamsAllPreserved() {
+        // globalParams=[global_only=global], commandParams=[start_only=start], varPool=[upstream_only=upstream]
+        final Property globalParam = Property.builder()
+                .prop("global_only")
+                .direct(Direct.IN)
+                .type(DataType.VARCHAR)
+                .value("global_val")
+                .build();
+        final Property commandParam = Property.builder()
+                .prop("start_only")
+                .direct(Direct.IN)
+                .type(DataType.VARCHAR)
+                .value("start_val")
+                .build();
+        final Property upstreamOutParam = Property.builder()
+                .prop("upstream_only")
+                .direct(Direct.OUT)
+                .type(DataType.VARCHAR)
+                .value("upstream_val")
+                .build();
+
+        final List<Property> result = SubWorkflowLogicTask.mergeParams(Arrays.asList(
+                Collections.singletonList(globalParam),
+                Collections.singletonList(commandParam),
+                Collections.singletonList(upstreamOutParam)));
+
+        assertThat(result).hasSize(3);
+        // Verify each parameter is present with its expected value
+        assertThat(
+                result.stream().anyMatch(p -> "global_only".equals(p.getProp()) && "global_val".equals(p.getValue())))
+                        .isTrue();
+        assertThat(result.stream().anyMatch(p -> "start_only".equals(p.getProp()) && "start_val".equals(p.getValue())))
+                .isTrue();
+        assertThat(
+                result.stream()
+                        .anyMatch(p -> "upstream_only".equals(p.getProp()) && "upstream_val".equals(p.getValue())))
+                                .isTrue();
+    }
+}

dolphinscheduler-master/src/test/java/org/apache/dolphinscheduler/server/master/engine/task/execution/TaskExecutionContextBuilderTest.java

@@ -0,0 +1,127 @@
+/*
+ * 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.server.master.engine.task.execution;
+
+import static com.google.common.truth.Truth.assertThat;
+
+import org.apache.dolphinscheduler.dao.entity.TaskInstance;
+import org.apache.dolphinscheduler.plugin.task.api.TaskExecutionContext;
+import org.apache.dolphinscheduler.plugin.task.api.enums.DataType;
+import org.apache.dolphinscheduler.plugin.task.api.enums.Direct;
+import org.apache.dolphinscheduler.plugin.task.api.model.Property;
+import org.apache.dolphinscheduler.plugin.task.api.utils.VarPoolUtils;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+
+import org.junit.jupiter.api.DisplayName;
+import org.junit.jupiter.api.Test;
+
+class TaskExecutionContextBuilderTest {
+
+    private static final String WORKFLOW_INSTANCE_HOST = "127.0.0.1:5678";
+
+    @Test
+    @DisplayName("Test that VarPool from TaskInstance is propagated to TaskExecutionContext")
+    void testBuildTaskInstanceRelatedInfo_copiesVarPool() {
+        // Given: a TaskInstance with a predecessor-scoped VarPool containing
+        // an OUT parameter from an upstream task
+        final Property upstreamOutParam = Property.builder()
+                .prop("output1")
+                .direct(Direct.OUT)
+                .type(DataType.VARCHAR)
+                .value("upstream_value")
+                .build();
+        final List<Property> predecessorScopedVarPool = Collections.singletonList(upstreamOutParam);
+
+        final TaskInstance taskInstance = new TaskInstance();
+        taskInstance.setId(1);
+        taskInstance.setName("sub_workflow_task");
+        taskInstance.setVarPool(VarPoolUtils.serializeVarPool(predecessorScopedVarPool));
+
+        // When: building the TaskExecutionContext
+        final TaskExecutionContext taskExecutionContext = TaskExecutionContextBuilder.get()
+                .buildTaskInstanceRelatedInfo(taskInstance)
+                .buildWorkflowInstanceHost(WORKFLOW_INSTANCE_HOST)
+                .create();
+
+        // Then: the VarPool is correctly propagated from TaskInstance to TaskExecutionContext
+        assertThat(taskExecutionContext.getVarPool()).isNotNull();
+        assertThat(taskExecutionContext.getVarPool()).hasSize(1);
+        final Property varPoolEntry = taskExecutionContext.getVarPool().get(0);
+        assertThat(varPoolEntry.getProp()).isEqualTo("output1");
+        assertThat(varPoolEntry.getDirect()).isEqualTo(Direct.OUT);
+        assertThat(varPoolEntry.getValue()).isEqualTo("upstream_value");
+    }
+
+    @Test
+    @DisplayName("Test that null VarPool in TaskInstance results in empty list in TaskExecutionContext")
+    void testBuildTaskInstanceRelatedInfo_nullVarPool() {
+        // Given: a TaskInstance with null VarPool
+        final TaskInstance taskInstance = new TaskInstance();
+        taskInstance.setId(1);
+        taskInstance.setName("sub_workflow_task");
+        taskInstance.setVarPool(null);
+
+        // When: building the TaskExecutionContext
+        final TaskExecutionContext taskExecutionContext = TaskExecutionContextBuilder.get()
+                .buildTaskInstanceRelatedInfo(taskInstance)
+                .buildWorkflowInstanceHost(WORKFLOW_INSTANCE_HOST)
+                .create();
+
+        // Then: the VarPool is an empty list (not null)
+        assertThat(taskExecutionContext.getVarPool()).isNotNull();
+        assertThat(taskExecutionContext.getVarPool()).isEmpty();
+    }
+
+    @Test
+    @DisplayName("Test that multiple OUT parameters from predecessor are all propagated")
+    void testBuildTaskInstanceRelatedInfo_multipleVarPoolEntries() {
+        // Given: a TaskInstance with multiple predecessor-scoped VarPool entries
+        final Property outParam1 = Property.builder()
+                .prop("output1")
+                .direct(Direct.OUT)
+                .type(DataType.VARCHAR)
+                .value("value1")
+                .build();
+        final Property outParam2 = Property.builder()
+                .prop("output2")
+                .direct(Direct.OUT)
+                .type(DataType.INTEGER)
+                .value("42")
+                .build();
+        final List<Property> predecessorScopedVarPool = Arrays.asList(outParam1, outParam2);
+
+        final TaskInstance taskInstance = new TaskInstance();
+        taskInstance.setId(1);
+        taskInstance.setName("sub_workflow_task");
+        taskInstance.setVarPool(VarPoolUtils.serializeVarPool(predecessorScopedVarPool));
+
+        // When: building the TaskExecutionContext
+        final TaskExecutionContext taskExecutionContext = TaskExecutionContextBuilder.get()
+                .buildTaskInstanceRelatedInfo(taskInstance)
+                .buildWorkflowInstanceHost(WORKFLOW_INSTANCE_HOST)
+                .create();
+
+        // Then: all VarPool entries are propagated
+        assertThat(taskExecutionContext.getVarPool()).hasSize(2);
+        assertThat(taskExecutionContext.getVarPool().get(0).getProp()).isEqualTo("output1");
+        assertThat(taskExecutionContext.getVarPool().get(1).getProp()).isEqualTo("output2");
+    }
+}