natas_hw
/
dolphinscheduler
0
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Merge 0d35482c44 into a13eacd61c

This commit is contained in:
zuo 2024-06-12 13:53:44 +08:00 committed by GitHub
parent a13eacd61c 0d35482c44
commit 0e69d33fba
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
7 changed files with 95 additions and 54 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

15
dolphinscheduler-task-plugin/dolphinscheduler-task-api/src/main/java/org/apache/dolphinscheduler/plugin/task/api/AbstractTask.java
View File

@ -26,7 +26,6 @@ import java.util.Map;
import java.util.StringJoiner;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
import lombok.Getter;
import lombok.Setter;
@ -35,10 +34,6 @@ import lombok.extern.slf4j.Slf4j;
@Slf4j
public abstract class AbstractTask {
private static String groupName1 = "paramName1";
private static String groupName2 = "paramName2";
public String rgex = String.format("['\"]\\$\\{(?<%s>.*?)}['\"]|\\$\\{(?<%s>.*?)}", groupName1, groupName2);
@Getter
@Setter
protected Map<String, String> taskOutputParams;
@ -173,24 +168,22 @@ public abstract class AbstractTask {
* regular expressions match the contents between two specified strings
*
* @param content content
* @param rgex rgex
* @param sqlParamsMap sql params map
* @param paramsPropsMap params props map
*/
public void setSqlParamsMap(String content, String rgex, Map<Integer, Property> sqlParamsMap,
public void setSqlParamsMap(String content, Map<Integer, Property> sqlParamsMap,
Map<String, Property> paramsPropsMap, int taskInstanceId) {
if (paramsPropsMap == null) {
return;
}
Pattern pattern = Pattern.compile(rgex);
Matcher m = pattern.matcher(content);
Matcher m = TaskConstants.SQL_PARAMS_PATTERN.matcher(content);
int index = 1;
while (m.find()) {
String paramName = m.group(groupName1);
String paramName = m.group(TaskConstants.GROUP_NAME1);
if (paramName == null) {
paramName = m.group(groupName2);
paramName = m.group(TaskConstants.GROUP_NAME2);
}
Property prop = paramsPropsMap.get(paramName);

10
dolphinscheduler-task-plugin/dolphinscheduler-task-api/src/main/java/org/apache/dolphinscheduler/plugin/task/api/TaskConstants.java
View File

@ -21,6 +21,7 @@ import org.apache.dolphinscheduler.common.constants.DateConstants;
import java.time.Duration;
import java.util.Set;
import java.util.regex.Pattern;
import com.google.common.collect.Sets;
@ -439,4 +440,13 @@ public class TaskConstants {
// Loop task constants
public static final Duration DEFAULT_LOOP_STATUS_INTERVAL = Duration.ofSeconds(5L);
/**
* sql params regex
*/
public static final String GROUP_NAME1 = "paramName1";
public static final String GROUP_NAME2 = "paramName2";
public static final String SQL_PARAMS_REGEX =
String.format("['\"]\\$\\{(?<%s>.*?)}['\"]|\\$\\{(?<%s>.*?)}", GROUP_NAME1, GROUP_NAME2);
public static final Pattern SQL_PARAMS_PATTERN = Pattern.compile(SQL_PARAMS_REGEX);
}

33
dolphinscheduler-task-plugin/dolphinscheduler-task-api/src/main/java/org/apache/dolphinscheduler/plugin/task/api/utils/ParameterUtils.java
View File

@ -23,6 +23,7 @@ import static org.apache.dolphinscheduler.plugin.task.api.TaskConstants.PARAMETE
import org.apache.dolphinscheduler.common.utils.DateUtils;
import org.apache.dolphinscheduler.common.utils.JSONUtils;
import org.apache.dolphinscheduler.plugin.task.api.TaskConstants;
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;
@ -192,24 +193,27 @@ public class ParameterUtils {
if (params == null || params.isEmpty()) {
return sql;
}
String[] split = sql.split("\\?");
if (split.length == 0) {
return sql;
}
StringBuilder ret = new StringBuilder(split[0]);
StringBuilder ret = new StringBuilder(sql);
Matcher m = TaskConstants.SQL_PARAMS_PATTERN.matcher(sql);
int index = 1;
for (int i = 1; i < split.length; i++) {
Property property = params.get(i);
int paramsIndex = 1;
// When matching with a regex, determine whether the corresponding property is a list.
while (m.find()) {
Property property = params.get(paramsIndex++);
if (property == null) {
continue;
}
String value = property.getValue();
StringBuilder tempReplace = new StringBuilder();
if (DataType.LIST.equals(property.getType())) {
List<Object> valueList = JSONUtils.toList(value, Object.class);
if (valueList.isEmpty() && StringUtils.isNotBlank(value)) {
valueList.add(value);
}
for (int j = 0; j < valueList.size(); j++) {
ret.append(PARAM_REPLACE_CHAR);
tempReplace.append(PARAM_REPLACE_CHAR);
if (j != valueList.size() - 1) {
ret.append(",");
tempReplace.append(",");
}
}
for (Object v : valueList) {
@ -231,14 +235,12 @@ public class ParameterUtils {
expandMap.put(index++, newProperty);
}
} else {
ret.append(PARAM_REPLACE_CHAR);
tempReplace.append(PARAM_REPLACE_CHAR);
expandMap.put(index++, property);
}
ret.append(split[i]);
}
if (PARAM_REPLACE_CHAR == sql.charAt(sql.length() - 1)) {
ret.append(PARAM_REPLACE_CHAR);
expandMap.put(index, params.get(split.length));
ret.replace(m.start(), m.end(), tempReplace.toString());
// After replacement, the string length will change, so a reset is required
m.reset(ret.toString());
}
params.clear();
params.putAll(expandMap);
@ -341,5 +343,4 @@ public class ParameterUtils {
}
return userDefParamsMaps;
}
}

20
dolphinscheduler-task-plugin/dolphinscheduler-task-api/src/test/java/org/apache/dolphinscheduler/plugin/task/api/utils/ParameterUtilsTest.java
View File

@ -23,6 +23,7 @@ import org.apache.dolphinscheduler.common.constants.DateConstants;
import org.apache.dolphinscheduler.common.utils.DateUtils;
import org.apache.dolphinscheduler.common.utils.JSONUtils;
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.parser.PlaceholderUtils;
@ -40,21 +41,25 @@ public class ParameterUtilsTest {
@Test
public void expandListParameter() {
Map<Integer, Property> params = new HashMap<>();
params.put(1,
new Property(null, null, DataType.LIST, JSONUtils.toJsonString(Lists.newArrayList("c1", "c2", "c3"))));
params.put(2, new Property(null, null, DataType.DATE, "2020-06-30"));
params.put(3, new Property(null, null, DataType.LIST,
new Property("col1", Direct.IN, DataType.LIST,
JSONUtils.toJsonString(Lists.newArrayList("c1", "c2", "c3"))));
params.put(2, new Property("date", Direct.IN, DataType.DATE, "2020-06-30"));
params.put(3, new Property("col2", Direct.IN, DataType.LIST,
JSONUtils.toJsonString(Lists.newArrayList(3.1415, 2.44, 3.44))));
String sql = ParameterUtils.expandListParameter(params,
"select * from test where col1 in (?) and date=? and col2 in (?)");
"select * from test where col1 in ('${col1}') and date='${date}' and col2 in ('${col2}')");
Assertions.assertEquals("select * from test where col1 in (?,?,?) and date=? and col2 in (?,?,?)", sql);
Assertions.assertEquals(7, params.size());
Map<Integer, Property> params2 = new HashMap<>();
params2.put(1, new Property(null, null, DataType.LIST, JSONUtils.toJsonString(Lists.newArrayList("c1"))));
params2.put(2, new Property(null, null, DataType.DATE, "2020-06-30"));
String sql2 = ParameterUtils.expandListParameter(params2, "select * from test where col1 in (?) and date=?");
params2.put(1,
new Property("col1", Direct.IN, DataType.LIST, JSONUtils.toJsonString(Lists.newArrayList("c1"))));
params2.put(2, new Property("date", Direct.IN, DataType.DATE, "2020-06-30"));
String sql2 = ParameterUtils.expandListParameter(params2,
"select * from test where col1 in ('${col}') and date='${date}'");
Assertions.assertEquals("select * from test where col1 in (?) and date=?", sql2);
Assertions.assertEquals(2, params2.size());
@ -113,4 +118,5 @@ public class ParameterUtilsTest {
Assertions.assertEquals("test Parameter", ParameterUtils.handleEscapes("test Parameter"));
Assertions.assertEquals("////%test////%Parameter", ParameterUtils.handleEscapes("%test%Parameter"));
}
}

5
dolphinscheduler-task-plugin/dolphinscheduler-task-procedure/src/main/java/org/apache/dolphinscheduler/plugin/task/procedure/ProcedureTask.java
View File

@ -26,6 +26,7 @@ import org.apache.dolphinscheduler.plugin.datasource.api.plugin.DataSourceClient
import org.apache.dolphinscheduler.plugin.datasource.api.plugin.DataSourceProcessorProvider;
import org.apache.dolphinscheduler.plugin.task.api.AbstractTask;
import org.apache.dolphinscheduler.plugin.task.api.TaskCallBack;
import org.apache.dolphinscheduler.plugin.task.api.TaskConstants;
import org.apache.dolphinscheduler.plugin.task.api.TaskException;
import org.apache.dolphinscheduler.plugin.task.api.TaskExecutionContext;
import org.apache.dolphinscheduler.plugin.task.api.enums.DataType;
@ -131,9 +132,9 @@ public class ProcedureTask extends AbstractTask {
}
private String formatSql(Map<Integer, Property> sqlParamsMap, Map<String, Property> paramsMap) {
setSqlParamsMap(procedureParameters.getMethod(), rgex, sqlParamsMap, paramsMap,
setSqlParamsMap(procedureParameters.getMethod(), sqlParamsMap, paramsMap,
taskExecutionContext.getTaskInstanceId());
return procedureParameters.getMethod().replaceAll(rgex, "?");
return procedureParameters.getMethod().replaceAll(TaskConstants.SQL_PARAMS_REGEX, "?");
}
/**

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

@ -449,17 +449,16 @@ public class SqlTask extends AbstractTask {
}
// special characters need to be escaped, ${} needs to be escaped
setSqlParamsMap(sql, rgex, sqlParamsMap, paramsMap, taskExecutionContext.getTaskInstanceId());
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
String formatSql = sql.replaceAll(rgex, "?");
// Convert the list parameter
formatSql = ParameterUtils.expandListParameter(sqlParamsMap, formatSql);
String formatSql = ParameterUtils.expandListParameter(sqlParamsMap, sql);
sqlBuilder.append(formatSql);
// print replace sql
printReplacedSql(sql, formatSql, rgex, sqlParamsMap);
printReplacedSql(sql, formatSql, TaskConstants.SQL_PARAMS_REGEX, sqlParamsMap);
return new SqlBinds(sqlBuilder.toString(), sqlParamsMap);
}

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

@ -17,6 +17,8 @@
package org.apache.dolphinscheduler.plugin.task.sql;
import org.apache.dolphinscheduler.common.utils.JSONUtils;
import org.apache.dolphinscheduler.plugin.task.api.TaskConstants;
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;
@ -24,6 +26,7 @@ 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.resource.DataSourceParameters;
import org.apache.dolphinscheduler.plugin.task.api.parameters.resource.ResourceParametersHelper;
import org.apache.dolphinscheduler.plugin.task.api.utils.ParameterUtils;
import org.apache.dolphinscheduler.spi.enums.DbType;
import java.util.HashMap;
@ -33,6 +36,8 @@ import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import com.google.common.collect.Lists;
class SqlTaskTest {
private SqlTask sqlTask;
@ -57,21 +62,21 @@ class SqlTaskTest {
void testReplacingSqlWithoutParams() {
String querySql = "select 1";
String expected = "select 1";
Assertions.assertEquals(expected, querySql.replaceAll(sqlTask.rgex, "?"));
Assertions.assertEquals(expected, querySql.replaceAll(TaskConstants.SQL_PARAMS_REGEX, "?"));
}
@Test
void testReplacingSqlWithDollarSymbol() {
String querySql = "select concat(amount, '$') as price from product";
String expected = "select concat(amount, '$') as price from product";
Assertions.assertEquals(expected, querySql.replaceAll(sqlTask.rgex, "?"));
Assertions.assertEquals(expected, querySql.replaceAll(TaskConstants.SQL_PARAMS_REGEX, "?"));
}
@Test
void testReplacingHiveLoadSql() {
String hiveLoadSql = "load inpath '/tmp/test_table/dt=${dt}' into table test_table partition(dt=${dt})";
String expected = "load inpath '/tmp/test_table/dt=?' into table test_table partition(dt=?)";
Assertions.assertEquals(expected, hiveLoadSql.replaceAll(sqlTask.rgex, "?"));
Assertions.assertEquals(expected, hiveLoadSql.replaceAll(TaskConstants.SQL_PARAMS_REGEX, "?"));
Map<Integer, Property> sqlParamsMap = new HashMap<>();
Map<Integer, Property> expectedSQLParamsMap = new HashMap<>();
@ -79,7 +84,7 @@ class SqlTaskTest {
expectedSQLParamsMap.put(2, new Property("dt", Direct.IN, DataType.VARCHAR, "1970"));
Map<String, Property> paramsMap = new HashMap<>();
paramsMap.put("dt", new Property("dt", Direct.IN, DataType.VARCHAR, "1970"));
sqlTask.setSqlParamsMap(hiveLoadSql, sqlTask.rgex, sqlParamsMap, paramsMap, 1);
sqlTask.setSqlParamsMap(hiveLoadSql, sqlParamsMap, paramsMap, 1);
Assertions.assertEquals(sqlParamsMap, expectedSQLParamsMap);
}
@ -87,38 +92,38 @@ class SqlTaskTest {
void testReplacingSelectSql() {
String querySql = "select id from student where dt='${dt}'";
String expected = "select id from student where dt=?";
Assertions.assertEquals(expected, querySql.replaceAll(sqlTask.rgex, "?"));
Assertions.assertEquals(expected, querySql.replaceAll(TaskConstants.SQL_PARAMS_REGEX, "?"));
Map<Integer, Property> sqlParamsMap = new HashMap<>();
Map<Integer, Property> expectedSQLParamsMap = new HashMap<>();
expectedSQLParamsMap.put(1, new Property("dt", Direct.IN, DataType.VARCHAR, "1970"));
Map<String, Property> paramsMap = new HashMap<>();
paramsMap.put("dt", new Property("dt", Direct.IN, DataType.VARCHAR, "1970"));
sqlTask.setSqlParamsMap(querySql, sqlTask.rgex, sqlParamsMap, paramsMap, 1);
sqlTask.setSqlParamsMap(querySql, sqlParamsMap, paramsMap, 1);
Assertions.assertEquals(sqlParamsMap, expectedSQLParamsMap);
querySql = "select id from student where dt=\"${dt}\"";
expected = "select id from student where dt=?";
Assertions.assertEquals(expected, querySql.replaceAll(sqlTask.rgex, "?"));
Assertions.assertEquals(expected, querySql.replaceAll(TaskConstants.SQL_PARAMS_REGEX, "?"));
sqlParamsMap.clear();
sqlTask.setSqlParamsMap(querySql, sqlTask.rgex, sqlParamsMap, paramsMap, 1);
sqlTask.setSqlParamsMap(querySql, sqlParamsMap, paramsMap, 1);
Assertions.assertEquals(sqlParamsMap, expectedSQLParamsMap);
querySql = "select id from student where dt=${dt}";
expected = "select id from student where dt=?";
Assertions.assertEquals(expected, querySql.replaceAll(sqlTask.rgex, "?"));
Assertions.assertEquals(expected, querySql.replaceAll(TaskConstants.SQL_PARAMS_REGEX, "?"));
sqlParamsMap.clear();
sqlTask.setSqlParamsMap(querySql, sqlTask.rgex, sqlParamsMap, paramsMap, 1);
sqlTask.setSqlParamsMap(querySql, sqlParamsMap, paramsMap, 1);
Assertions.assertEquals(sqlParamsMap, expectedSQLParamsMap);
querySql = "select id from student where dt=${dt} and gender=1";
expected = "select id from student where dt=? and gender=1";
Assertions.assertEquals(expected, querySql.replaceAll(sqlTask.rgex, "?"));
Assertions.assertEquals(expected, querySql.replaceAll(TaskConstants.SQL_PARAMS_REGEX, "?"));
sqlParamsMap.clear();
sqlTask.setSqlParamsMap(querySql, sqlTask.rgex, sqlParamsMap, paramsMap, 1);
sqlTask.setSqlParamsMap(querySql, sqlParamsMap, paramsMap, 1);
Assertions.assertEquals(sqlParamsMap, expectedSQLParamsMap);
}
@ -126,7 +131,7 @@ class SqlTaskTest {
void testReplacingSqlNonGreedy() {
String querySql = "select id from student where year=${year} and month=${month} and gender=1";
String expected = "select id from student where year=? and month=? and gender=1";
Assertions.assertEquals(expected, querySql.replaceAll(sqlTask.rgex, "?"));
Assertions.assertEquals(expected, querySql.replaceAll(TaskConstants.SQL_PARAMS_REGEX, "?"));
Map<Integer, Property> sqlParamsMap = new HashMap<>();
Map<Integer, Property> expectedSQLParamsMap = new HashMap<>();
@ -135,11 +140,37 @@ class SqlTaskTest {
Map<String, Property> paramsMap = new HashMap<>();
paramsMap.put("year", new Property("year", Direct.IN, DataType.VARCHAR, "1970"));
paramsMap.put("month", new Property("month", Direct.IN, DataType.VARCHAR, "12"));
sqlTask.setSqlParamsMap(querySql, sqlTask.rgex, sqlParamsMap, paramsMap, 1);
sqlTask.setSqlParamsMap(querySql, sqlParamsMap, paramsMap, 1);
Assertions.assertEquals(sqlParamsMap, expectedSQLParamsMap);
}
@Test
void splitSql() {
}
@Test
void testReplacingSqlHasQuestionMarkAndParams() {
String querySql =
"select id, concat('?', year) from student where year=${year} and month=${month} and gender in ('${gender}')";
String expected =
"select id, concat('?', year) from student where year=? and month=? and gender in (?,?)";
Map<Integer, Property> sqlParamsMap = new HashMap<>();
Map<Integer, Property> expectedSQLParamsMap = new HashMap<>();
expectedSQLParamsMap.put(1, new Property("year", Direct.IN, DataType.VARCHAR, "1970"));
expectedSQLParamsMap.put(2, new Property("month", Direct.IN, DataType.VARCHAR, "12"));
expectedSQLParamsMap.put(3,
new Property("gender", Direct.IN, DataType.LIST, JSONUtils.toJsonString(Lists.newArrayList(1, 2))));
Map<String, Property> paramsMap = new HashMap<>();
paramsMap.put("year", new Property("year", Direct.IN, DataType.VARCHAR, "1970"));
paramsMap.put("month", new Property("month", Direct.IN, DataType.VARCHAR, "12"));
paramsMap.put("gender",
new Property("gender", Direct.IN, DataType.LIST, JSONUtils.toJsonString(Lists.newArrayList(1, 2))));
sqlTask.setSqlParamsMap(querySql, sqlParamsMap, paramsMap, 1);
Assertions.assertEquals(sqlParamsMap, expectedSQLParamsMap);
String formatSql = ParameterUtils.expandListParameter(sqlParamsMap, querySql);
Assertions.assertEquals(4, sqlParamsMap.size());
Assertions.assertEquals(expected, formatSql);
}
}