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

docs/docs/en/guide/task/sql.md

@@ -27,6 +27,15 @@ 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 replaces custom-parameter placeholders in the SQL text before submitting it to the datasource through JDBC `Statement`.
+
+- Both `${name}` and `!{name}` insert the parameter value as text. DolphinScheduler does not add quotes, escape characters, interpret the parameter type, or otherwise adapt the value to a database dialect.
+- Write datasource-specific SQL syntax, including any required quotes, in the SQL statement or parameter value. Only use trusted parameter values because text replacement does not prevent SQL injection.
+- Rendering scans the source SQL once. Placeholder-like text introduced by a parameter value is not rendered again.
+- If a source placeholder is unclosed or refers to a missing custom parameter, the SQL task fails before submitting the statement to the datasource.
+
 ## Task Example
 
 ### Hive Table Create Example

docs/docs/zh/guide/task/sql.md

@@ -27,6 +27,15 @@ SQL任务类型，用于连接数据库并执行相应SQL。
 - 前置sql:前置sql在sql语句之前执行。
 - 后置sql:后置sql在sql语句之后执行。
 
+## 参数渲染
+
+SQL 任务会先在 SQL 文本中替换自定义参数占位符，然后通过 JDBC `Statement` 提交到数据源执行。
+
+- `${name}` 和 `!{name}` 都会将参数值作为文本插入。DolphinScheduler 不会自动加引号、转义字符、解释参数类型，也不会按数据库方言转换参数值。
+- 请在 SQL 语句或参数值中编写数据源所需的 SQL 语法，包括必要的引号。文本替换无法防止 SQL 注入，因此只能使用可信的参数值。
+- 渲染只扫描一次源 SQL。参数值中包含的占位符形式文本不会再次渲染。
+- 如果源 SQL 中的占位符未闭合或引用了不存在的自定义参数，SQL 任务会在向数据源提交语句前失败。
+
 ## 任务样例
 
 ### Hive表创建示例

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

@@ -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 {
 

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

@@ -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;
@@ -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;
@@ -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);
         }
     }
@@ -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());
             }
@@ -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() {
@@ -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()));
@@ -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());
     }
 
 }

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

@@ -0,0 +1,81 @@
+/*
+ * 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.plugin.task.sql;
+
+import org.apache.dolphinscheduler.plugin.task.api.TaskException;
+import org.apache.dolphinscheduler.plugin.task.api.model.Property;
+
+import org.apache.commons.lang3.StringUtils;
+
+import java.util.Map;
+
+final class SqlTaskParameterRenderer {
+
+    private SqlTaskParameterRenderer() {
+    }
+
+    static String render(String sql, Map<String, Property> paramsMap, int taskInstanceId) {
+        if (StringUtils.isEmpty(sql)) {
+            return sql;
+        }
+        return renderParameters(sql, paramsMap, taskInstanceId);
+    }
+
+    private static String renderParameters(String sql, Map<String, Property> paramsMap, int taskInstanceId) {
+        StringBuilder renderedSql = new StringBuilder(sql.length());
+        int index = 0;
+        while (index < sql.length()) {
+            char marker = sql.charAt(index);
+            if (!isPlaceholderStart(sql, index, marker)) {
+                renderedSql.append(marker);
+                index++;
+                continue;
+            }
+
+            int end = sql.indexOf('}', index + 2);
+            if (end < 0) {
+                throw new TaskException(String.format(
+                        "Unclosed SQL parameter placeholder in task instance with id: %s",
+                        taskInstanceId));
+            }
+
+            String paramName = sql.substring(index + 2, end);
+            Property property = getProperty(paramsMap, paramName, taskInstanceId);
+            renderedSql.append(StringUtils.defaultString(property.getValue()));
+            index = end + 1;
+        }
+        return renderedSql.toString();
+    }
+
+    private static boolean isPlaceholderStart(String sql, int index, char marker) {
+        return (marker == '$' || marker == '!')
+                && index + 1 < sql.length()
+                && sql.charAt(index + 1) == '{';
+    }
+
+    private static Property getProperty(Map<String, Property> paramsMap, String paramName, int taskInstanceId) {
+        Property property = paramsMap == null ? null : paramsMap.get(paramName);
+        if (property == null) {
+            throw new TaskException(String.format(
+                    "No Property with paramName: %s is found in paramsMap of task instance with id: %s",
+                    paramName,
+                    taskInstanceId));
+        }
+        return property;
+    }
+}