This is an automated email from the ASF dual-hosted git repository.
wenjun 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 0deafba12c fix procedure task param pass (#11919)
0deafba12c is described below
commit 0deafba12c839bbfc3b49090ff5b5cdec8021984
Author: caishunfeng <[email protected]>
AuthorDate: Wed Sep 14 13:44:01 2022 +0800
fix procedure task param pass (#11919)
---
docs/docs/en/guide/task/stored-procedure.md | 48 +++++++++++++--------
docs/docs/zh/guide/task/stored-procedure.md | 19 ++++++--
docs/img/procedure-en.png | Bin 165706 -> 0 bytes
docs/img/procedure_edit.png | Bin 49986 -> 115800 bytes
.../plugin/task/api/AbstractTask.java | 32 +++++++-------
.../plugin/task/procedure/ProcedureTask.java | 40 +++++++++++------
6 files changed, 89 insertions(+), 50 deletions(-)
diff --git a/docs/docs/en/guide/task/stored-procedure.md
b/docs/docs/en/guide/task/stored-procedure.md
index 5aff75207c..a1a3942091 100644
--- a/docs/docs/en/guide/task/stored-procedure.md
+++ b/docs/docs/en/guide/task/stored-procedure.md
@@ -2,27 +2,39 @@
- Execute the stored procedure according to the selected DataSource.
-> Drag from the toolbar
task node into the canvas, as shown in the figure below:
+> Drag from the `PROCEDURE` task node into the canvas, as shown in the figure
below:
<p align="center">
- <img src="../../../../img/procedure-en.png" width="80%" />
+ <img src="../../../../img/procedure_edit.png" width="80%" />
</p>
## Task Parameters
-| **Parameter** | **Description** |
-| ------- | ---------- |
-| Node Name | Set the name of the task. Node names within a workflow
definition are unique. |
-| Run flag | Indicates whether the node can be scheduled normally. If it is
not necessary to execute, you can turn on the prohibiting execution switch. |
-| Description | Describes the function of this node. |
-| Task priority | When the number of worker threads is insufficient, they are
executed in order from high to low according to the priority, and they are
executed according to the first-in, first-out principle when the priority is
the same. |
-| Worker group | The task is assigned to the machines in the worker group for
execution. If Default is selected, a worker machine will be randomly selected
for execution. |
-| Task group name | The group in Resources, if not configured, it will not be
used. |
-| Environment Name | Configure the environment in which to run the script. |
-| Number of failed retries | The number of times the task is resubmitted after
failure. It supports drop-down and manual filling. |
-| Failure Retry Interval | The time interval for resubmitting the task if the
task fails. It supports drop-down and manual filling. |
-| Timeout alarm | Check Timeout Alarm and Timeout Failure. When the task
exceeds the "timeout duration", an alarm email will be sent and the task
execution will fail. |
-| DataSource | The DataSource type of the stored procedure supports MySQL and
POSTGRESQL, select the corresponding DataSource. |
-| Method | The method name of the stored procedure. |
-| Custom parameters | The custom parameter types of the stored procedure
support `IN` and `OUT`, and the data types support: VARCHAR, INTEGER, LONG,
FLOAT, DOUBLE, DATE, TIME, TIMESTAMP and BOOLEAN. |
-| Predecessor task | Selecting the predecessor task of the current task will
set the selected predecessor task as the upstream of the current task. |
\ No newline at end of file
+| **Parameter** |
**Description**
|
+|--------------------------|-------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
+| Node Name | Set the name of the task. Node names within a
workflow definition are unique.
|
+| Run flag | Indicates whether the node can be scheduled
normally. If it is not necessary to execute, you can turn on the prohibiting
execution switch.
|
+| Description | Describes the function of this node.
|
+| Task priority | When the number of worker threads is
insufficient, they are executed in order from high to low according to the
priority, and they are executed according to the first-in, first-out principle
when the priority is the same. |
+| Worker group | The task is assigned to the machines in the
worker group for execution. If Default is selected, a worker machine will be
randomly selected for execution.
|
+| Task group name | The group in Resources, if not configured, it
will not be used.
|
+| Environment Name | Configure the environment in which to run the
script.
|
+| Number of failed retries | The number of times the task is resubmitted after
failure. It supports drop-down and manual filling.
|
+| Failure Retry Interval | The time interval for resubmitting the task if
the task fails. It supports drop-down and manual filling.
|
+| Timeout alarm | Check Timeout Alarm and Timeout Failure. When the
task exceeds the "timeout duration", an alarm email will be sent and the task
execution will fail.
|
+| DataSource | The DataSource type of the stored procedure
supports MySQL, POSTGRESQL, ORACLE.
|
+| SQL Statement | call a stored procedure, such as `call
test(${in1},${out1});`.
|
+| Custom parameters | The custom parameter types of the stored
procedure support `IN` and `OUT`, and the data types support: VARCHAR, INTEGER,
LONG, FLOAT, DOUBLE, DATE, TIME, TIMESTAMP and BOOLEAN.
|
+| Predecessor task | Selecting the predecessor task of the current
task will set the selected predecessor task as the upstream of the current
task.
|
+
+## Remark
+
+- Prepare: Create a stored procedure in the database, such as:
+
+ ```
+ CREATE PROCEDURE dolphinscheduler.test(in in1 INT, out out1 INT)
+ begin
+ set out1=in1;
+ END
+ ```
+
diff --git a/docs/docs/zh/guide/task/stored-procedure.md
b/docs/docs/zh/guide/task/stored-procedure.md
index 48b3dc946b..b383ae3d10 100644
--- a/docs/docs/zh/guide/task/stored-procedure.md
+++ b/docs/docs/zh/guide/task/stored-procedure.md
@@ -1,12 +1,23 @@
# 存储过程节点
- 根据选择的数据源，执行存储过程。
->
拖动工具栏中的任务节点到画板中，如下图所示：
+
+> 拖动工具栏中的`PROCEDURE`任务节点到画板中，如下图所示：
<p align="center">
<img src="../../../../img/procedure_edit.png" width="80%" />
</p>
-- 数据源：存储过程的数据源类型支持MySQL和POSTGRESQL两种，选择对应的数据源
-- 方法：是存储过程的方法名称
--
自定义参数：存储过程的自定义参数类型支持IN、OUT两种，数据类型支持VARCHAR、INTEGER、LONG、FLOAT、DOUBLE、DATE、TIME、TIMESTAMP、BOOLEAN九种数据类型
+- 前提：在该数据库里面创建存储过程，如：
+
+```
+CREATE PROCEDURE dolphinscheduler.test(in in1 INT, out out1 INT)
+begin
+ set out1=in1;
+END
+```
+
+- 数据源：存储过程的数据源类型支持MySQL、POSTGRESQL、ORACLE，选择对应的数据源
+- SQL Statement：调用存储过程，如 `call test(${in1},${out1});`；
+-
自定义参数：存储过程的自定义参数类型支持IN、OUT两种，数据类型支持VARCHAR、INTEGER、LONG、FLOAT、DOUBLE、DATE、TIME、TIMESTAMP、BOOLEAN九种数据类型；
+
diff --git a/docs/img/procedure-en.png b/docs/img/procedure-en.png
deleted file mode 100644
index b4679b37ab..0000000000
Binary files a/docs/img/procedure-en.png and /dev/null differ
diff --git a/docs/img/procedure_edit.png b/docs/img/procedure_edit.png
index 004c0ba1b2..06bcd99d47 100644
Binary files a/docs/img/procedure_edit.png and b/docs/img/procedure_edit.png
differ
diff --git
a/dolphinscheduler-task-plugin/dolphinscheduler-task-api/src/main/java/org/apache/dolphinscheduler/plugin/task/api/AbstractTask.java
b/dolphinscheduler-task-plugin/dolphinscheduler-task-api/src/main/java/org/apache/dolphinscheduler/plugin/task/api/AbstractTask.java
index 47cc7d9c75..89b80a4f83 100644
---
a/dolphinscheduler-task-plugin/dolphinscheduler-task-api/src/main/java/org/apache/dolphinscheduler/plugin/task/api/AbstractTask.java
+++
b/dolphinscheduler-task-plugin/dolphinscheduler-task-api/src/main/java/org/apache/dolphinscheduler/plugin/task/api/AbstractTask.java
@@ -21,17 +21,8 @@ import
org.apache.dolphinscheduler.plugin.task.api.enums.TaskExecutionStatus;
import org.apache.dolphinscheduler.plugin.task.api.model.Property;
import org.apache.dolphinscheduler.plugin.task.api.model.TaskAlertInfo;
import
org.apache.dolphinscheduler.plugin.task.api.parameters.AbstractParameters;
-import org.apache.dolphinscheduler.spi.utils.StringUtils;
-
-import java.io.BufferedReader;
-import java.io.File;
-import java.io.FileInputStream;
-import java.io.IOException;
-import java.io.InputStreamReader;
-import java.nio.charset.StandardCharsets;
-import java.util.HashSet;
+
import java.util.Map;
-import java.util.Set;
import java.util.StringJoiner;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.regex.Matcher;
@@ -49,7 +40,8 @@ public abstract class AbstractTask {
public static final Marker FINALIZE_SESSION_MARKER =
MarkerFactory.getMarker("FINALIZE_SESSION");
- protected final Logger logger =
LoggerFactory.getLogger(String.format(TaskConstants.TASK_LOG_LOGGER_NAME_FORMAT,
getClass()));
+ protected final Logger logger =
+
LoggerFactory.getLogger(String.format(TaskConstants.TASK_LOG_LOGGER_NAME_FORMAT,
getClass()));
public String rgex = "['\"]*\\$\\{(.*?)\\}['\"]*";
@@ -226,7 +218,11 @@ public abstract class AbstractTask {
* @param paramsPropsMap params props map
*/
public void setSqlParamsMap(String content, String rgex, Map<Integer,
Property> sqlParamsMap,
- Map<String, Property> paramsPropsMap,int
taskInstanceId) {
+ Map<String, Property> paramsPropsMap, int
taskInstanceId) {
+ if (paramsPropsMap == null) {
+ return;
+ }
+
Pattern pattern = Pattern.compile(rgex);
Matcher m = pattern.matcher(content);
int index = 1;
@@ -236,14 +232,18 @@ public abstract class AbstractTask {
Property prop = paramsPropsMap.get(paramName);
if (prop == null) {
- logger.error("setSqlParamsMap: No Property with paramName: {}
is found in paramsPropsMap of task instance"
- + " with id: {}. So couldn't put Property in
sqlParamsMap.", paramName, taskInstanceId);
+ logger.error(
+ "setSqlParamsMap: No Property with paramName: {} is
found in paramsPropsMap of task instance"
+ + " with id: {}. So couldn't put Property in
sqlParamsMap.",
+ paramName, taskInstanceId);
} else {
sqlParamsMap.put(index, prop);
index++;
- logger.info("setSqlParamsMap: Property with paramName: {} put
in sqlParamsMap of content {} successfully.", paramName, content);
+ logger.info(
+ "setSqlParamsMap: Property with paramName: {} put in
sqlParamsMap of content {} successfully.",
+ paramName, content);
}
}
}
-}
\ No newline at end of file
+}
diff --git
a/dolphinscheduler-task-plugin/dolphinscheduler-task-procedure/src/main/java/org/apache/dolphinscheduler/plugin/task/procedure/ProcedureTask.java
b/dolphinscheduler-task-plugin/dolphinscheduler-task-procedure/src/main/java/org/apache/dolphinscheduler/plugin/task/procedure/ProcedureTask.java
index 743b936e22..19890dd90a 100644
---
a/dolphinscheduler-task-plugin/dolphinscheduler-task-procedure/src/main/java/org/apache/dolphinscheduler/plugin/task/procedure/ProcedureTask.java
+++
b/dolphinscheduler-task-plugin/dolphinscheduler-task-procedure/src/main/java/org/apache/dolphinscheduler/plugin/task/procedure/ProcedureTask.java
@@ -17,6 +17,9 @@
package org.apache.dolphinscheduler.plugin.task.procedure;
+import static
org.apache.dolphinscheduler.plugin.task.api.TaskConstants.EXIT_CODE_FAILURE;
+import static
org.apache.dolphinscheduler.plugin.task.api.TaskConstants.EXIT_CODE_SUCCESS;
+
import
org.apache.dolphinscheduler.plugin.datasource.api.plugin.DataSourceClientProvider;
import org.apache.dolphinscheduler.plugin.datasource.api.utils.DataSourceUtils;
import org.apache.dolphinscheduler.plugin.task.api.AbstractTask;
@@ -42,8 +45,7 @@ import java.sql.Types;
import java.util.HashMap;
import java.util.Map;
-import static
org.apache.dolphinscheduler.plugin.task.api.TaskConstants.EXIT_CODE_FAILURE;
-import static
org.apache.dolphinscheduler.plugin.task.api.TaskConstants.EXIT_CODE_SUCCESS;
+import com.google.common.collect.Maps;
/**
* procedure task
@@ -74,14 +76,16 @@ public class ProcedureTask extends AbstractTask {
logger.info("procedure task params {}",
taskExecutionContext.getTaskParams());
- this.procedureParameters =
JSONUtils.parseObject(taskExecutionContext.getTaskParams(),
ProcedureParameters.class);
+ this.procedureParameters =
+ JSONUtils.parseObject(taskExecutionContext.getTaskParams(),
ProcedureParameters.class);
// check parameters
if (!procedureParameters.checkParameters()) {
throw new RuntimeException("procedure task params is not valid");
}
- procedureTaskExecutionContext =
procedureParameters.generateExtendedContext(taskExecutionContext.getResourceParametersHelper());
+ procedureTaskExecutionContext =
+
procedureParameters.generateExtendedContext(taskExecutionContext.getResourceParametersHelper());
}
@Override
@@ -98,13 +102,22 @@ public class ProcedureTask extends AbstractTask {
// load class
DbType dbType = DbType.valueOf(procedureParameters.getType());
// get datasource
- ConnectionParam connectionParam =
DataSourceUtils.buildConnectionParams(DbType.valueOf(procedureParameters.getType()),
- procedureTaskExecutionContext.getConnectionParams());
+ ConnectionParam connectionParam =
+
DataSourceUtils.buildConnectionParams(DbType.valueOf(procedureParameters.getType()),
+
procedureTaskExecutionContext.getConnectionParams());
// get jdbc connection
connection =
DataSourceClientProvider.getInstance().getConnection(dbType, connectionParam);
+
Map<Integer, Property> sqlParamsMap = new HashMap<>();
- Map<String, Property> paramsMap =
taskExecutionContext.getPrepareParamsMap();
+ Map<String, Property> paramsMap =
taskExecutionContext.getPrepareParamsMap() == null ? Maps.newHashMap()
+ : taskExecutionContext.getPrepareParamsMap();
+ if (procedureParameters.getOutProperty() != null) {
+ // set out params before format sql
+ paramsMap.putAll(procedureParameters.getOutProperty());
+ }
+
+ // format sql
String proceduerSql = formatSql(sqlParamsMap, paramsMap);
// call method
stmt = connection.prepareCall(proceduerSql);
@@ -137,7 +150,8 @@ public class ProcedureTask extends AbstractTask {
private String formatSql(Map<Integer, Property> sqlParamsMap, Map<String,
Property> paramsMap) {
// combining local and global parameters
- setSqlParamsMap(procedureParameters.getMethod(), rgex, sqlParamsMap,
paramsMap, taskExecutionContext.getTaskInstanceId());
+ setSqlParamsMap(procedureParameters.getMethod(), rgex, sqlParamsMap,
paramsMap,
+ taskExecutionContext.getTaskInstanceId());
return procedureParameters.getMethod().replaceAll(rgex, "?");
}
@@ -168,8 +182,8 @@ public class ProcedureTask extends AbstractTask {
* @return outParameterMap
* @throws Exception Exception
*/
- private Map<Integer, Property> getOutParameterMap(CallableStatement stmt,
Map<Integer, Property> paramsMap
- , Map<String, Property> totalParamsMap) throws Exception {
+ private Map<Integer, Property> getOutParameterMap(CallableStatement stmt,
Map<Integer, Property> paramsMap,
+ Map<String, Property>
totalParamsMap) throws Exception {
Map<Integer, Property> outParameterMap = new HashMap<>();
if (procedureParameters.getLocalParametersMap() == null) {
return outParameterMap;
@@ -180,7 +194,8 @@ public class ProcedureTask extends AbstractTask {
for (Map.Entry<Integer, Property> entry : paramsMap.entrySet()) {
Property property = entry.getValue();
if (property.getDirect().equals(Direct.IN)) {
- ParameterUtils.setInParameter(index, stmt,
property.getType(), totalParamsMap.get(property.getProp()).getValue());
+ ParameterUtils.setInParameter(index, stmt,
property.getType(),
+ totalParamsMap.get(property.getProp()).getValue());
} else if (property.getDirect().equals(Direct.OUT)) {
setOutParameter(index, stmt, property.getType(),
totalParamsMap.get(property.getProp()).getValue());
outParameterMap.put(index, property);
@@ -237,7 +252,8 @@ public class ProcedureTask extends AbstractTask {
* @param dataType dataType
* @throws SQLException SQLException
*/
- private Object getOutputParameter(CallableStatement stmt, int index,
String prop, DataType dataType) throws SQLException {
+ private Object getOutputParameter(CallableStatement stmt, int index,
String prop,
+ DataType dataType) throws SQLException {
Object value = null;
switch (dataType) {
case VARCHAR:
