Skip to content

[DSIP-89][SqlTask] Avoid using PrepareStatement in SqlTask #18398

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
qiuyanjun888 wants to merge 2 commits into
base: dev
Choose a base branch
from
Open

[DSIP-89][SqlTask] Avoid using PrepareStatement in SqlTask #18398

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
f927607
[DSIP-17263][Task] Render SQL parameters before statement execution
qiuyanjun888 Jul 6, 2026
2c0cf5d
Merge branch 'dev' into DSIP-17263
SbloodyS Jul 10, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
10 changes: 10 additions & 0 deletions docs/docs/en/guide/task/sql.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -27,6 +27,16 @@ Refer to [datasource-setting](../installation/datasource-setting.md) `DataSource
| Pre-SQL | Pre-SQL executes before the SQL statement. |
| Post-SQL | Post-SQL executes after the SQL statement. |

## Parameter Rendering

SQL task renders custom parameters in the SQL text before submitting it to the datasource through JDBC `Statement`.

- `${name}` renders a SQL literal by default: string/date/time/timestamp values are wrapped in single quotes and embedded single quotes are escaped, while numeric and boolean values are rendered without quotes.
- `LIST` values render as comma-separated SQL literals, for example `id in (${ids})` can become `id in (1,'a')`.
- When `${name}` is part of an identifier, for example `create table test_${bizDate}`, the value is rendered as an identifier fragment without extra quotes.
- `!{name}` keeps raw text replacement semantics for SQL fragments that must not be quoted automatically. Use it only for trusted SQL fragments because DolphinScheduler does not escape raw text replacements.
- If a placeholder refers to a missing custom parameter, the SQL task fails fast instead of leaving the placeholder unresolved.

## Task Example

### Hive Table Create Example
Expand Down
10 changes: 10 additions & 0 deletions docs/docs/zh/guide/task/sql.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -27,6 +27,16 @@ SQL任务类型，用于连接数据库并执行相应SQL。
- 前置sql:前置sql在sql语句之前执行。
- 后置sql:后置sql在sql语句之后执行。

## 参数渲染

SQL 任务会先在 SQL 文本中渲染自定义参数，然后通过 JDBC `Statement` 提交到数据源执行。

- `${name}` 默认渲染为 SQL 字面量：字符串、日期、时间、时间戳会使用单引号包裹并转义内部单引号，数字和布尔值不额外加引号。
- `LIST` 类型会渲染为逗号分隔的 SQL 字面量，例如 `id in (${ids})` 可以渲染为 `id in (1,'a')`。
- 当 `${name}` 是标识符的一部分时，例如 `create table test_${bizDate}`，参数值会作为标识符片段直接渲染，不额外加引号。
- `!{name}` 保留原始文本替换语义，适用于不希望自动加引号的 SQL 片段。该替换不会由 DolphinScheduler 转义，请仅用于可信 SQL 片段。
- 如果占位符引用了不存在的自定义参数，SQL 任务会快速失败，而不是保留未解析的占位符继续执行。

## 任务样例

### Hive表创建示例
Expand Down
2 changes: 1 addition & 1 deletion ...cheduler-task-sql/src/main/java/org/apache/dolphinscheduler/plugin/task/sql/SqlBinds.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -22,7 +22,7 @@
import java.util.Map;

/**
* Used to contains both prepared sql string and its to-be-bind parameters
* Used to contain the rendered SQL string and legacy parameter map.
*/
public class SqlBinds {

Expand Down
93 changes: 12 additions & 81 deletions ...scheduler-task-sql/src/main/java/org/apache/dolphinscheduler/plugin/task/sql/SqlTask.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -47,7 +47,6 @@
import java.nio.file.Files;
import java.nio.file.Paths;
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.ResultSetMetaData;
import java.sql.SQLException;
Expand All @@ -60,8 +59,6 @@
import java.util.Map;
import java.util.Optional;
import java.util.Set;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
import java.util.stream.Collectors;

import lombok.extern.slf4j.Slf4j;
Expand Down Expand Up @@ -335,9 +332,9 @@ private void sendAttachment(int groupId, String title, String content) {
}

private String executeQuery(Connection connection, SqlBinds sqlBinds, String handlerType) throws Exception {
try (PreparedStatement statement = prepareStatementAndBind(connection, sqlBinds)) {
try (Statement statement = createStatement(connection)) {
log.info("{} statement execute query, for sql: {}", handlerType, sqlBinds.getSql());
ResultSet resultSet = statement.executeQuery();
ResultSet resultSet = statement.executeQuery(sqlBinds.getSql());
return resultProcess(resultSet);
}
}
Expand All @@ -346,8 +343,8 @@ private String executeUpdate(Connection connection, List<SqlBinds> statementsBin
String handlerType) throws Exception {
int result = 0;
for (SqlBinds sqlBind : statementsBinds) {
try (PreparedStatement tmpStatement = prepareStatementAndBind(connection, sqlBind)) {
result = tmpStatement.executeUpdate();
try (Statement tmpStatement = createStatement(connection)) {
result = tmpStatement.executeUpdate(sqlBind.getSql());
log.info("{} statement execute update result: {}, for sql: {}", handlerType, result,
sqlBind.getSql());
}
Expand All @@ -356,61 +353,27 @@ private String executeUpdate(Connection connection, List<SqlBinds> statementsBin
}

/**
* preparedStatement bind
* create statement
*
* @param connection connection
* @param sqlBinds sqlBinds
* @return PreparedStatement
* @throws Exception Exception
* @return Statement
*/
private PreparedStatement prepareStatementAndBind(Connection connection, SqlBinds sqlBinds) {
private Statement createStatement(Connection connection) {
// is the timeout set
// todo: we need control the timeout at master side.
boolean timeoutFlag = taskExecutionContext.getTaskTimeoutStrategy() == TaskTimeoutStrategy.FAILED
|| taskExecutionContext.getTaskTimeoutStrategy() == TaskTimeoutStrategy.WARNFAILED;
try {
PreparedStatement stmt = connection.prepareStatement(sqlBinds.getSql());
Statement stmt = connection.createStatement();
if (timeoutFlag) {
stmt.setQueryTimeout(taskExecutionContext.getTaskTimeout());
}
stmt.setMaxRows(sqlParameters.getLimit() <= 0 ? QUERY_LIMIT : sqlParameters.getLimit());
Map<Integer, Property> params = sqlBinds.getParamsMap();
if (params != null) {
for (Map.Entry<Integer, Property> entry : params.entrySet()) {
Property prop = entry.getValue();
ParameterUtils.setInParameter(entry.getKey(), stmt, prop.getType(), prop.getValue());
}
}
log.info("prepare statement replace sql : {}, sql parameters : {}", sqlBinds.getSql(),
sqlBinds.getParamsMap());
sessionStatement = stmt;
return stmt;
} catch (Exception exception) {
throw new TaskException("SQL task prepareStatementAndBind error", exception);
}
}

/**
* print replace sql
*
* @param content content
* @param formatSql format sql
* @param rgex rgex
* @param sqlParamsMap sql params map
*/
private void printReplacedSql(String content, String formatSql, String rgex, Map<Integer, Property> sqlParamsMap) {
// parameter print style
log.info("after replace sql , preparing : {}", formatSql);
StringBuilder logPrint = new StringBuilder("replaced sql , parameters:");
if (sqlParamsMap == null) {
log.info("printReplacedSql: sqlParamsMap is null.");
} else {
for (int i = 1; i <= sqlParamsMap.size(); i++) {
logPrint.append(sqlParamsMap.get(i).getValue()).append("(").append(sqlParamsMap.get(i).getType())
.append(")");
}
throw new TaskException("SQL task createStatement error", exception);
}
log.info("Sql Params are {}", logPrint);
}

private void ensureSqlContent() {
Expand Down Expand Up @@ -445,9 +408,6 @@ private void ensureSqlContent() {
* @return SqlBinds
*/
private SqlBinds getSqlAndSqlParamsMap(String sql) {
Map<Integer, Property> sqlParamsMap = new HashMap<>();
StringBuilder sqlBuilder = new StringBuilder();
// new
// replace variable TIME with $[YYYYmmddd...] in sql when history run job and batch complement job
sql = ParameterUtils.replaceScheduleTime(sql,
DateUtils.timeStampToDate(taskExecutionContext.getScheduleTime()));
Expand All @@ -464,38 +424,9 @@ private SqlBinds getSqlAndSqlParamsMap(String sql) {
sqlParameters.setTitle(title);
}

// spell SQL according to the final user-defined variable
if (paramsMap == null) {
sqlBuilder.append(sql);
return new SqlBinds(sqlBuilder.toString(), sqlParamsMap);
}

// special characters need to be escaped, ${} needs to be escaped
setSqlParamsMap(sql, sqlParamsMap, paramsMap, taskExecutionContext.getTaskInstanceId());
// Replace the original value in sql ！{...} ，Does not participate in precompilation
String rgexo = "['\"]*\\!\\{(.*?)\\}['\"]*";
sql = replaceOriginalValue(sql, rgexo, paramsMap);
// replace the ${} of the SQL statement with the Placeholder
// Convert the list parameter
String formatSql = ParameterUtils.expandListParameter(sqlParamsMap, sql);
sqlBuilder.append(formatSql);
// print replace sql
printReplacedSql(sql, formatSql, TaskConstants.SQL_PARAMS_REGEX, sqlParamsMap);
return new SqlBinds(sqlBuilder.toString(), sqlParamsMap);
}

private String replaceOriginalValue(String content, String rgex, Map<String, Property> sqlParamsMap) {
Pattern pattern = Pattern.compile(rgex);
while (true) {
Matcher m = pattern.matcher(content);
if (!m.find()) {
break;
}
String paramName = m.group(1);
String paramValue = sqlParamsMap.get(paramName).getValue();
content = m.replaceFirst(paramValue);
}
return content;
String renderedSql = SqlTaskParameterRenderer.render(sql, paramsMap, taskExecutionContext.getTaskInstanceId());
log.info("rendered sql : {}", renderedSql);
return new SqlBinds(renderedSql, Collections.emptyMap());
}

}
Loading
Loading