时间增量抽取，格式化时间字符串空格被替换成%的问题修复；顺便修复一下DB2按照OWNERTYPE获取schemas不准确的问题，修改为直接排除系统schemas

LockiJiang · LockiJiang · commit a665f406de20 · 2020-09-15T14:06:05.000+08:00
diff --git a/datax-admin/src/main/java/com/wugui/datax/admin/tool/meta/DB2DatabaseMeta.java b/datax-admin/src/main/java/com/wugui/datax/admin/tool/meta/DB2DatabaseMeta.java
@@ -47,7 +47,7 @@ public String getSQLQueryColumns(String... args) {
 
     @Override
     public String getSQLQueryTableSchema(String... args) {
-        return "SELECT SCHEMANAME FROM SYSCAT.SCHEMATA WHERE OWNERTYPE = 'U'";
+        return "SELECT SCHEMANAME FROM SYSCAT.SCHEMATA WHERE SCHEMANAME NOT IN('NULLID','ROOT','SQLJ','SYSCAT','SYSFUN','SYSIBM','SYSIBMADM','SYSIBMINTERNAL','SYSIBMTS','SYSPROC','SYSPUBLIC','SYSSTAT','SYSTOOLS')";
     }
 
     @Override
diff --git a/datax-executor/src/main/java/com/wugui/datax/executor/service/command/BuildCommand.java b/datax-executor/src/main/java/com/wugui/datax/executor/service/command/BuildCommand.java
@@ -10,10 +10,7 @@
 
 import java.io.File;
 import java.text.SimpleDateFormat;
-import java.util.ArrayList;
-import java.util.Arrays;
-import java.util.Date;
-import java.util.List;
+import java.util.*;
 
 import static com.wugui.datatx.core.util.Constants.SPLIT_COMMA;
 import static com.wugui.datax.executor.service.jobhandler.DataXConstant.*;
@@ -53,54 +50,64 @@ public static String[] buildDataXExecutorCmd(TriggerParam tgParam, String tmpFil
     private static String buildDataXParam(TriggerParam tgParam) {
         StringBuilder doc = new StringBuilder();
         String jvmParam = StringUtils.isNotBlank(tgParam.getJvmParam()) ? tgParam.getJvmParam().trim() : tgParam.getJvmParam();
-        String partitionStr = tgParam.getPartitionInfo();
         if (StringUtils.isNotBlank(jvmParam)) {
             doc.append(JVM_CM).append(TRANSFORM_QUOTES).append(jvmParam).append(TRANSFORM_QUOTES);
         }
+        return doc.toString();
+    }
 
+    public static HashMap<String, String> buildDataXParamToMap(TriggerParam tgParam) {
+        String partitionStr = tgParam.getPartitionInfo();
         Integer incrementType = tgParam.getIncrementType();
         String replaceParam = StringUtils.isNotBlank(tgParam.getReplaceParam()) ? tgParam.getReplaceParam().trim() : null;
-
         if (incrementType != null && replaceParam != null) {
-
-            if (IncrementTypeEnum.TIME.getCode() == incrementType) {
-                if (doc.length() > 0) doc.append(SPLIT_SPACE);
+            if (IncrementTypeEnum.ID.getCode() == incrementType) {
+                long startId = tgParam.getStartId();
+                long endId = tgParam.getEndId();
+                String formatParam = String.format(replaceParam, startId, endId);
+                return getKeyValue(formatParam);
+            } else if (IncrementTypeEnum.TIME.getCode() == incrementType) {
                 String replaceParamType = tgParam.getReplaceParamType();
-
-                if (StringUtils.isBlank(replaceParamType) || replaceParamType.equals("Timestamp")) {
+                if (StringUtils.isBlank(replaceParamType) || "Timestamp".equals(replaceParamType)) {
                     long startTime = tgParam.getStartTime().getTime() / 1000;
                     long endTime = tgParam.getTriggerTime().getTime() / 1000;
-                    doc.append(PARAMS_CM).append(TRANSFORM_QUOTES).append(String.format(replaceParam, startTime, endTime));
+                    String formatParam = String.format(replaceParam, startTime, endTime);
+                    return getKeyValue(formatParam);
                 } else {
                     SimpleDateFormat sdf = new SimpleDateFormat(replaceParamType);
-                    String endTime = sdf.format(tgParam.getTriggerTime()).replaceAll(SPLIT_SPACE, PERCENT);
-                    String startTime = sdf.format(tgParam.getStartTime()).replaceAll(SPLIT_SPACE, PERCENT);
-                    doc.append(PARAMS_CM).append(TRANSFORM_QUOTES).append(String.format(replaceParam, startTime, endTime));
-                }
-                //buildPartitionCM(doc, partitionStr);
-                doc.append(TRANSFORM_QUOTES);
+                    String endTime = sdf.format(tgParam.getTriggerTime());
+                    String startTime = sdf.format(tgParam.getStartTime());
 
-            } else if (IncrementTypeEnum.ID.getCode() == incrementType) {
-                long startId = tgParam.getStartId();
-                long endId = tgParam.getEndId();
-                if (doc.length() > 0) doc.append(SPLIT_SPACE);
-                doc.append(PARAMS_CM).append(TRANSFORM_QUOTES).append(String.format(replaceParam, startId, endId));
-                doc.append(TRANSFORM_QUOTES);
+                    String formatParam = String.format(replaceParam, startTime, endTime);
+                    return getKeyValue(formatParam);
+                }
             }
         }
 
         if (incrementType != null && IncrementTypeEnum.PARTITION.getCode() == incrementType) {
             if (StringUtils.isNotBlank(partitionStr)) {
                 List<String> partitionInfo = Arrays.asList(partitionStr.split(SPLIT_COMMA));
-                if (doc.length() > 0) doc.append(SPLIT_SPACE);
-                doc.append(PARAMS_CM).append(TRANSFORM_QUOTES).append(String.format(PARAMS_CM_V_PT, buildPartition(partitionInfo))).append(TRANSFORM_QUOTES);
+                String formatParam = String.format(PARAMS_CM_V_PT, buildPartition(partitionInfo));
+                return getKeyValue(formatParam);
             }
-        }
 
-        JobLogger.log("------------------Command parameters:" + doc);
-        return doc.toString();
+        }
+        return null;
     }
 
+    private static HashMap<String, String> getKeyValue(String formatParam) {
+        String[] paramArr = formatParam.split(PARAMS_SYSTEM);
+        HashMap<String, String> map = new HashMap<String, String>();
+
+        for (String param : paramArr) {
+            if (StringUtils.isNotBlank(param)) {
+                param = param.trim();
+                String[] keyValue = param.split(PARAMS_EQUALS);
+                map.put(keyValue[0], keyValue[1]);
+            }
+        }
+        return map;
+    }
 
     private void buildPartitionCM(StringBuilder doc, String partitionStr) {
         if (StringUtils.isNotBlank(partitionStr)) {
diff --git a/datax-executor/src/main/java/com/wugui/datax/executor/service/jobhandler/DataXConstant.java b/datax-executor/src/main/java/com/wugui/datax/executor/service/jobhandler/DataXConstant.java
@@ -19,6 +19,9 @@ public class DataXConstant {
 
     public static final String PARAMS_CM = "-p";
 
+    public static final String PARAMS_SYSTEM = "-D";
+    public static final String PARAMS_EQUALS = "=";
+
     public static final String PARAMS_CM_V_PT = "-Dpartition=%s";
 
     public static final String DEFAULT_JSON = "jsons";
diff --git a/datax-executor/src/main/java/com/wugui/datax/executor/service/jobhandler/ExecutorJobHandler.java b/datax-executor/src/main/java/com/wugui/datax/executor/service/jobhandler/ExecutorJobHandler.java
@@ -17,9 +17,14 @@
 import org.springframework.stereotype.Component;
 
 import java.io.*;
+import java.util.HashMap;
+import java.util.Map;
 import java.util.concurrent.FutureTask;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
 
 import static com.wugui.datax.executor.service.command.BuildCommand.buildDataXExecutorCmd;
+import static com.wugui.datax.executor.service.command.BuildCommand.buildDataXParamToMap;
 import static com.wugui.datax.executor.service.jobhandler.DataXConstant.DEFAULT_JSON;
 import static com.wugui.datax.executor.service.logparse.AnalysisStatistics.analysisStatisticsLog;
 
@@ -39,6 +44,7 @@ public class ExecutorJobHandler extends IJobHandler {
     @Value("${datax.pypath}")
     private String dataXPyPath;
 
+    private  static final Pattern VARIABLE_PATTERN = Pattern.compile("(\\$)\\{?(\\w+)\\}?");
 
     @Override
     public ReturnT<String> execute(TriggerParam trigger) {
@@ -47,8 +53,12 @@ public ReturnT<String> execute(TriggerParam trigger) {
         Thread errThread = null;
         String tmpFilePath;
         LogStatistics logStatistics = null;
+
+        HashMap<String, String> keyValueMap = buildDataXParamToMap(trigger);
+        String jobJson = replaceVariable(trigger.getJobJson(),keyValueMap);
+
         //Generate JSON temporary file
-        tmpFilePath = generateTemJsonFile(trigger.getJobJson());
+        tmpFilePath = generateTemJsonFile(jobJson);
 
         try {
             String[] cmdarrayFinal = buildDataXExecutorCmd(trigger, tmpFilePath,dataXPyPath);
@@ -97,7 +107,25 @@ public ReturnT<String> execute(TriggerParam trigger) {
         }
     }
 
+    public static String replaceVariable(final String param,HashMap<String, String> variableMap) {
+        Map<String, String> mapping = new HashMap<String, String>();
 
+        Matcher matcher = VARIABLE_PATTERN.matcher(param);
+        while (matcher.find()) {
+            String variable = matcher.group(2);
+            String value = variableMap.get(variable);
+            if (StringUtils.isBlank(value)) {
+                value = matcher.group();
+            }
+            mapping.put(matcher.group(), value);
+        }
+
+        String retString = param;
+        for (final String key : mapping.keySet()) {
+            retString = retString.replace(key, mapping.get(key));
+        }
+        return retString;
+    }
 
     private String generateTemJsonFile(String jobJson) {
         String tmpFilePath;

Original file line number	Diff line number	Diff line change
`@@ -47,7 +47,7 @@ public String getSQLQueryColumns(String... args) {`
`47`	`47`
`48`	`48`	`@Override`
`49`	`49`	`public String getSQLQueryTableSchema(String... args) {`
`50`		`- return "SELECT SCHEMANAME FROM SYSCAT.SCHEMATA WHERE OWNERTYPE = 'U'";`
	`50`	`+ return "SELECT SCHEMANAME FROM SYSCAT.SCHEMATA WHERE SCHEMANAME NOT IN('NULLID','ROOT','SQLJ','SYSCAT','SYSFUN','SYSIBM','SYSIBMADM','SYSIBMINTERNAL','SYSIBMTS','SYSPROC','SYSPUBLIC','SYSSTAT','SYSTOOLS')";`
`51`	`51`	`}`
`52`	`52`
`53`	`53`	`@Override`