fix sql varPool not work

Author: Zzih96

dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/executor/LogicTaskExecutor.java

@@ -59,10 +59,6 @@ protected void doTriggerTaskPlugin() {
         logicTask.start();
     }
 
-    @Override
-    protected void closureTaskContext() {
-    }
-
     @SneakyThrows
     @Override
     public void pause() {

dolphinscheduler-task-executor/src/main/java/org/apache/dolphinscheduler/task/executor/AbstractTaskExecutor.java

@@ -79,8 +79,6 @@ public void start() {
         }
 
         doTriggerTaskPlugin();
-
-        closureTaskContext();
     }
 
     @Override
@@ -127,8 +125,6 @@ protected void transitTaskExecutorState(final TaskExecutorState taskExecutorStat
 
     protected abstract void doTriggerTaskPlugin();
 
-    protected abstract void closureTaskContext();
-
     protected void initializeTaskContext() {
         taskExecutionContext.setStartTime(System.currentTimeMillis());
         log.info(TaskLogMarkers.excludeInTaskLog(), "Initialized taskContext {}",

dolphinscheduler-task-plugin/dolphinscheduler-task-sql/src/main/java/org/apache/dolphinscheduler/plugin/task/sql/SqlTask.java

@@ -198,6 +198,9 @@ public void executeFuncAndSql(List<SqlBinds> mainStatementsBinds,
 
             // post execute
             executeUpdate(connection, postStatementsBinds, "post");
+
+            // set varPool
+            taskExecutionContext.setVarPool(sqlParameters.getVarPool());
         } catch (Exception e) {
             log.error("execute sql error: {}", e.getMessage());
             throw e;

dolphinscheduler-task-plugin/dolphinscheduler-task-sql/src/test/java/org/apache/dolphinscheduler/plugin/task/sql/SqlTaskTest.java

@@ -24,6 +24,7 @@
 import org.apache.dolphinscheduler.plugin.task.api.enums.Direct;
 import org.apache.dolphinscheduler.plugin.task.api.enums.ResourceType;
 import org.apache.dolphinscheduler.plugin.task.api.model.Property;
+import org.apache.dolphinscheduler.plugin.task.api.parameters.SqlParameters;
 import org.apache.dolphinscheduler.plugin.task.api.parameters.resource.DataSourceParameters;
 import org.apache.dolphinscheduler.plugin.task.api.parameters.resource.ResourceParametersHelper;
 import org.apache.dolphinscheduler.plugin.task.api.utils.ParameterUtils;
@@ -173,4 +174,33 @@ void testReplacingSqlHasQuestionMarkAndParams() {
         Assertions.assertEquals(4, sqlParamsMap.size());
         Assertions.assertEquals(expected, formatSql);
     }
+
+    @Test
+    void testVarPoolSetting() {
+        // 创建包含OUT参数的SqlParameters
+        SqlParameters sqlParameters = new SqlParameters();
+        sqlParameters.setType("HIVE");
+        sqlParameters.setDatasource(1);
+        sqlParameters.setSql("select id, name from user where id = 1");
+
+        // 设置OUT参数
+        Property outParam = new Property("id", Direct.OUT, DataType.VARCHAR, "");
+        sqlParameters.setLocalParams(Lists.newArrayList(outParam));
+
+        // 模拟SQL执行结果
+        String sqlResult = "[{\"id\":\"1\",\"name\":\"test_user\"}]";
+
+        // 调用dealOutParam方法处理输出参数
+        sqlParameters.dealOutParam(sqlResult);
+
+        // 验证varPool是否正确设置
+        Assertions.assertNotNull(sqlParameters.getVarPool());
+        Assertions.assertEquals(1, sqlParameters.getVarPool().size());
+
+        // 验证参数值是否正确
+        Property varPoolParam = sqlParameters.getVarPool().get(0);
+        Assertions.assertEquals("id", varPoolParam.getProp());
+        Assertions.assertEquals("1", varPoolParam.getValue());
+        Assertions.assertEquals(Direct.OUT, varPoolParam.getDirect());
+    }
 }

dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/executor/PhysicalTaskExecutor.java

@@ -87,14 +87,6 @@ public void updateTaskInstanceInfo(final int taskInstanceId) {
         });
     }
 
-    @Override
-    protected void closureTaskContext() {
-        log.info("Begin to closure taskContext.");
-        taskExecutionContext.setVarPool(physicalTask.getParameters().getVarPool());
-        log.info("Set taskContext varPool {}", JSONUtils.toPrettyJsonString(taskExecutionContext.getVarPool()));
-        log.info("End closure taskContext.");
-    }
-
     @Override
     protected TaskExecutorState doTrackTaskPluginStatus() {
         return TaskExecutorStateMappings.mapState(physicalTask.getExitStatus());