COMMIT

src/main/java/com/jz/dmp/cmdexectool/scheduler/common/task/sync/SyncParameters.java

@@ -11,6 +11,7 @@ import com.jz.dmp.cmdexectool.mapper.DmpSyncingDatasourceDao;
 import com.jz.dmp.cmdexectool.scheduler.common.enums.MyDbType;
 import com.jz.dmp.cmdexectool.scheduler.common.process.ResourceInfo;
 import com.jz.dmp.cmdexectool.scheduler.common.task.AbstractParameters;
+import com.jz.dmp.cmdexectool.scheduler.common.utils.ParameterUtils;
 import com.jz.dmp.cmdexectool.scheduler.common.utils.StringUtils;
 import com.jz.dmp.cmdexectool.scheduler.dao.datasource.MyBaseDataSource;
 import org.springframework.ui.freemarker.FreeMarkerConfigurationFactoryBean;
@@ -158,7 +159,7 @@ public class SyncParameters extends AbstractParameters {
                 || targetTypeId == DatasouceTypeConstant.PostgreSQL
                 || targetTypeId == DatasouceTypeConstant.Informix
                 || targetTypeId == DatasouceTypeConstant.DB2)) {
-            getJdbcSink(targetDatasource, targetObj, freeMarkerConfig, publicKey, source_table_name);
+            getJdbcSink(targetDatasource, targetObj, freeMarkerConfig, publicKey, source_table_name, mappingObj);
         }
         if (targetTypeId == DatasouceTypeConstant.SFTP) {
             getSinkSftp(targetDatasource, publicKey, targetObj, freeMarkerConfig, source_table_name);
@@ -167,8 +168,8 @@ public class SyncParameters extends AbstractParameters {
             getSinkKudu(targetDatasource, freeMarkerConfig, source_table_name);
         }
         if (targetTypeId == DatasouceTypeConstant.Hive) {
-            //source
-            getSinkHive(freeMarkerConfig,sourceObj, source_table_name,envModel );
+            //sink
+            getSinkHive(freeMarkerConfig, sourceObj, source_table_name, envModel);
         }
         //transform
         if (mappingObj.size() > 0 && null != mappingObj) {
@@ -184,7 +185,7 @@ public class SyncParameters extends AbstractParameters {
         waterdropScript = FreeMarkerUtils.freemakerNoneWebJson(CommConstant.WATERDROP_FTL, waterdropModel, freeMarkerConfig);
     }
 
-    private void getSinkHive(FreeMarkerConfigurationFactoryBean freeMarkerConfig, Map<String, Object> sourceObj, String source_table_name,Map<String, String> envModel ) {
+    private void getSinkHive(FreeMarkerConfigurationFactoryBean freeMarkerConfig, Map<String, Object> sourceObj, String source_table_name, Map<String, String> envModel) {
         Map hiveModel = new HashMap();
         hiveModel.put("pre_sql", " select * from  " + source_table_name);
         hiveModel.put("result_table_name", source_table_name);
@@ -205,7 +206,7 @@ public class SyncParameters extends AbstractParameters {
         env = FreeMarkerUtils.freemakerNoneWebJson(CommConstant.WATERDROP_FTL_ENV, envModel, freeMarkerConfig);
     }
 
-    private void getSinkKudu(DmpSyncingDatasource targetDatasource,FreeMarkerConfigurationFactoryBean freeMarkerConfig, String source_table_name) {
+    private void getSinkKudu(DmpSyncingDatasource targetDatasource, FreeMarkerConfigurationFactoryBean freeMarkerConfig, String source_table_name) {
         Map kuduModel = new HashMap();
         kuduModel.put("kuduMaster", targetDatasource.getHost() + ":" + targetDatasource.getPort()); //主机名
         kuduModel.put("result_table_name", source_table_name); //spark生成的临时表名
@@ -220,7 +221,7 @@ public class SyncParameters extends AbstractParameters {
         source = source + FreeMarkerUtils.freemakerNoneWebJson(CommConstant.WATERDROP_FTL_SINK_KUDU, kuduModel, freeMarkerConfig);
     }
 
-    private void getSinkSftp(DmpSyncingDatasource targetDatasource, String publicKey, Map<String, Object> targetObj, FreeMarkerConfigurationFactoryBean freeMarkerConfig,String source_table_name) {
+    private void getSinkSftp(DmpSyncingDatasource targetDatasource, String publicKey, Map<String, Object> targetObj, FreeMarkerConfigurationFactoryBean freeMarkerConfig, String source_table_name) {
         Map sftpModel = new HashMap();
         sftpModel.put("host", targetDatasource.getHost()); //主机名
         sftpModel.put("user", targetDatasource.getUserName()); //用户
@@ -324,9 +325,10 @@ public class SyncParameters extends AbstractParameters {
     }
 
     //sink
-    public void getJdbcSink(DmpSyncingDatasource targetDatasource, Map<String, Object> targetObj, FreeMarkerConfigurationFactoryBean freeMarkerConfig, String publicKey, String source_table_name) {
+    public void getJdbcSink(DmpSyncingDatasource targetDatasource, Map<String, Object> targetObj, FreeMarkerConfigurationFactoryBean freeMarkerConfig, String publicKey, String source_table_name, List<Map<String, Object>> mappingObj) {
         String postImportStatement = String.valueOf(targetObj.get("postImportStatement")); //导入后语句
         String preImportStatement = String.valueOf(targetObj.get("preImportStatement")); //导入前语句
+        String targetTable = String.valueOf(targetObj.get("targetTable"));//目标表
         preStatements = new ArrayList<String>();
         preStatements.add(preImportStatement);
         posStatements = new ArrayList<String>();
@@ -340,20 +342,43 @@ public class SyncParameters extends AbstractParameters {
         targetBaseDataSource.setPassword(password);
         targetBaseDataSource.setMyDbType(myDbType);
 
+
+        String saveMode = "append";
+        String primaryKeyConflict = "";
+        if (StringUtils.isNotEmpty(String.valueOf(targetObj.get("primaryKeyConflict")))) {
+            primaryKeyConflict = String.valueOf(targetObj.get("primaryKeyConflict"));
+        }
+        if (myDbType.PostgreSQL == myDbType){
+            String pgSqlWriteMode = String.valueOf(targetObj.get("pgSqlWriteMode")); //pgsql 导入模式
+            if (StringUtils.isNotEmpty(pgSqlWriteMode)) {
+                saveMode = pgSqlWriteMode;
+            }
+        }
+
+        if (CommConstant.PRIMARY_KEY_CONFLICT_REPLACE.equals(primaryKeyConflict)
+                || CommConstant.PRIMARY_KEY_CONFLICT_UPDATE.equals(primaryKeyConflict)) {
+            saveMode = "update";
+        }
         Map jdbcSinkModel = new HashMap();
-        jdbcSinkModel.put("driver", targetDatasource.getDriverClassName());
-        jdbcSinkModel.put("url", targetDatasource.getJdbcUrl());
         //存储模式，支持overwrite、append、update、ignore、error
         //在save_mode指定为update时配置，用于指定键冲突的更新语句模板
-        jdbcSinkModel.put("save_mode", targetObj.get("primaryKeyConflict"));
+        jdbcSinkModel.put("save_mode", saveMode);
         //当存储模式是 overwrite时，仅清除表中数据
-        if (null != targetObj.get("primaryKeyConflict")) { //主键冲突
-            if ("overwrite".equals(targetObj.get("primaryKeyConflict"))) {
+        if (StringUtils.isNotEmpty(primaryKeyConflict)) { //主键冲突
+            if ("overwrite".equals(primaryKeyConflict)) {
                 jdbcSinkModel.put("truncate", "true");
             } else {
                 jdbcSinkModel.put("truncate", "false");
             }
         }
+        if ("update".equals(saveMode)) {
+            String customUpdateStmt = ParameterUtils.syncColumnMappingHandlerConflict(targetTable, mappingObj, myDbType, targetObj);
+            if (org.apache.commons.lang3.StringUtils.isNotEmpty(customUpdateStmt)) {
+                jdbcSinkModel.put("customUpdateStmt", customUpdateStmt);
+            }
+        }
+        jdbcSinkModel.put("driver", targetDatasource.getDriverClassName());
+        jdbcSinkModel.put("url", targetDatasource.getJdbcUrl());
         jdbcSinkModel.put("dbtable", targetObj.get("targetTable")); //目标表
         jdbcSinkModel.put("user", targetDatasource.getUserName());
         jdbcSinkModel.put("password", password);

src/main/java/com/jz/dmp/cmdexectool/scheduler/common/utils/ParameterUtils.java

@@ -259,24 +259,24 @@ public class ParameterUtils {
     }
 
     /**
-    * @Title: columnMappingHandler  
-    * @Description: TODO(开发任务SQL：字段映射处理)  
      * @param @param  jsonStr
      * @param @return 参数
-    * @return List<Map<String,String>>    返回类型  
+     * @return List<Map                                                                                                                               <                                                                                                                               String                                                                                                                               ,                                                                                                                               String>>    返回类型
      * @throws
+     * @Title: columnMappingHandler
+     * @Description: TODO(开发任务SQL ： 字段映射处理)
      */
     public static String columnMappingHandler(String jsonStr) {
         JSONObject jsonObject = JSONObject.parseObject(jsonStr);
         JSONArray sourceArray = jsonObject.getJSONArray("sourceFields");
         Map<String, JSONObject> sourceMap = new HashMap<String, JSONObject>();
-    	for (int index=0; index<sourceArray.size(); index++) {
+        for (int index = 0; index < sourceArray.size(); index++) {
             JSONObject sourceObj = sourceArray.getJSONObject(index);
             sourceMap.put(sourceObj.getString("customSoruceFieldId"), sourceObj);
         }
         JSONArray targetArray = jsonObject.getJSONArray("targetFields");
         Map<String, JSONObject> targetMap = new HashMap<String, JSONObject>();
-    	for (int index=0; index<targetArray.size(); index++) {
+        for (int index = 0; index < targetArray.size(); index++) {
             JSONObject targetObj = targetArray.getJSONObject(index);
             targetMap.put(targetObj.getString("customTargetFieldId"), targetObj);
         }
@@ -285,7 +285,7 @@ public class ParameterUtils {
 
         JSONArray mappingArray = jsonObject.getJSONArray("columnMapping");
         Integer size = mappingArray.size();
-    	for (int index=0; index<size; index++) {
+        for (int index = 0; index < size; index++) {
             JSONObject mappingObj = mappingArray.getJSONObject(index);
             JSONObject sourceObj = sourceMap.get(mappingObj.getString("customSoruceFieldId"));
             JSONObject targetObj = targetMap.get(mappingObj.getString("customTargetFieldId"));
@@ -297,10 +297,10 @@ public class ParameterUtils {
             String targetFieldName = targetObj.getString("targetFieldName");
             //String targetFieldType = targetObj.getString("targetFieldType");
 
-    		sb.append(sourceFieldName +" AS "+targetFieldName);
-    		if (index<size-1) {
+            sb.append(sourceFieldName + " AS " + targetFieldName);
+            if (index < size - 1) {
                 sb.append(", ");
-			}else {
+            } else {
                 sb.append(" ");
             }
         }
@@ -310,24 +310,24 @@ public class ParameterUtils {
     }
 
     /**
-     * @Title: columnMappingHandler  
-     * @Description: TODO(开发任务SQL：生成主键冲突语句)  
      * @param @param  jsonStr
      * @param @return 参数
-     * @return List<Map<String,String>>    返回类型  
+     * @return List<Map                                                                                                                               <                                                                                                                               String                                                                                                                               ,                                                                                                                               String>>    返回类型
      * @throws
+     * @Title: columnMappingHandler
+     * @Description: TODO(开发任务SQL ： 生成主键冲突语句)
      */
     public static String columnMappingHandlerConflict(String jsonStr, String tableName, MyDbType myDbType) {
         JSONObject jsonObject = JSONObject.parseObject(jsonStr);
         JSONArray sourceArray = jsonObject.getJSONArray("sourceFields");
         Map<String, JSONObject> sourceMap = new HashMap<String, JSONObject>();
-     	for (int index=0; index<sourceArray.size(); index++) {
+        for (int index = 0; index < sourceArray.size(); index++) {
             JSONObject sourceObj = sourceArray.getJSONObject(index);
             sourceMap.put(sourceObj.getString("customSoruceFieldId"), sourceObj);
         }
         JSONArray targetArray = jsonObject.getJSONArray("targetFields");
         Map<String, JSONObject> targetMap = new HashMap<String, JSONObject>();
-     	for (int index=0; index<targetArray.size(); index++) {
+        for (int index = 0; index < targetArray.size(); index++) {
             JSONObject targetObj = targetArray.getJSONObject(index);
             targetMap.put(targetObj.getString("customTargetFieldId"), targetObj);
         }
@@ -338,7 +338,7 @@ public class ParameterUtils {
 
         JSONArray mappingArray = jsonObject.getJSONArray("columnMapping");
         Integer size = mappingArray.size();
-     	for (int index=0; index<size; index++) {
+        for (int index = 0; index < size; index++) {
             JSONObject mappingObj = mappingArray.getJSONObject(index);
             JSONObject sourceObj = sourceMap.get(mappingObj.getString("customSoruceFieldId"));
             JSONObject targetObj = targetMap.get(mappingObj.getString("customTargetFieldId"));
@@ -354,11 +354,11 @@ public class ParameterUtils {
             //设置values语句
             valuesSqlSb.append("?");
 
-     		if (index!=size-1) {
+            if (index != size - 1) {
                 valuesSqlSb.append(",");
             }
 
-     		if ("0".equals(isPrimaryKey) && myDbType.MySQL==myDbType) {
+            if ("0".equals(isPrimaryKey) && myDbType.MySQL == myDbType) {
 
                 //设置update语句
                 updateSqlSb.append(targetFieldName);
@@ -367,11 +367,11 @@ public class ParameterUtils {
                 updateSqlSb.append(")");
                 updateSqlSb.append(",");
 
-			} else if (myDbType.PostgreSQL==myDbType) {
+            } else if (myDbType.PostgreSQL == myDbType) {
                 if ("1".equals(isPrimaryKey)) {
                     primaryKeySqlSb.append("targetFieldName");
                     primaryKeySqlSb.append(",");
-				}else {
+                } else {
                     //设置update语句
                     updateSqlSb.append(targetFieldName);
                     updateSqlSb.append("=EXCLUDED.");
@@ -383,27 +383,27 @@ public class ParameterUtils {
 
         String primaryKeySqlStr = "";
         if (org.apache.commons.lang3.StringUtils.isNotEmpty(primaryKeySqlSb)) {
-     		primaryKeySqlStr = primaryKeySqlSb.substring(0, primaryKeySqlSb.length()-1);
+            primaryKeySqlStr = primaryKeySqlSb.substring(0, primaryKeySqlSb.length() - 1);
         }
         String updateSqlStr = "";
         if (org.apache.commons.lang3.StringUtils.isNotEmpty(updateSqlSb)) {
-     		updateSqlStr = updateSqlSb.substring(0, updateSqlSb.length()-1);
+            updateSqlStr = updateSqlSb.substring(0, updateSqlSb.length() - 1);
         }
 
         StringBuilder sb = new StringBuilder();
 
-     	if (myDbType.MySQL==myDbType) {
+        if (myDbType.MySQL == myDbType) {
             sb.append("INSERT INTO ");
-         	sb.append(tableName+" ");
+            sb.append(tableName + " ");
             sb.append("VALUES(");
             sb.append(valuesSqlSb);
             sb.append(") ");
 
             sb.append("ON DUPLICATE KEY UPDATE ");
             sb.append(updateSqlStr);
-		} else if (myDbType.PostgreSQL==myDbType) {
+        } else if (myDbType.PostgreSQL == myDbType) {
             sb.append("INSERT INTO ");
-	     	sb.append(tableName+" ");
+            sb.append(tableName + " ");
             sb.append("VALUES(");
             sb.append(valuesSqlSb);
             sb.append(") ");
@@ -418,4 +418,92 @@ public class ParameterUtils {
         return sb.toString();
     }
 
+
+    /**
+     * @param @param  jsonStr
+     * @param @return 参数
+     * @throws
+     * @Title: columnMappingHandler
+     * @Description: TODO(开发任务SQL ： 生成主键冲突语句)
+     */
+    public static String syncColumnMappingHandlerConflict(String tableName, List<Map<String, Object>> mappingObj, MyDbType myDbType, Map<String, Object> targetObj) {
+        List<Map> targetColumn = (List<Map>) JSONObject.parse(String.valueOf(targetObj.get("column")));
+        StringBuilder valuesSqlSb = new StringBuilder();
+        StringBuilder primaryKeySqlSb = new StringBuilder();
+        StringBuilder updateSqlSb = new StringBuilder();
+
+        if (mappingObj.size() > 0 && mappingObj != null) {
+            for (Map<String, Object> mapping : mappingObj) {
+                String sourceFieldId = String.valueOf(mapping.get("sourceFieldId"));
+                String sourceFieldName = String.valueOf(mapping.get("sourceField"));
+                String targetFieldId = String.valueOf(mapping.get("targetFieldId"));
+                String targetFieldName = String.valueOf(mapping.get("targetField"));
+                String isPrimaryKey = "";
+                if (targetColumn.size() > 0) {
+                    for (Map targetStr : targetColumn) {
+                        String targetColumnId = String.valueOf(targetStr.get("customGraphId"));
+                        if (targetColumnId.equals(targetFieldId)) {
+                            isPrimaryKey = String.valueOf(targetStr.get("isPk"));
+                            break;
+                        }
+                    }
+                }
+                //设置values语句
+                valuesSqlSb.append("?");
+                valuesSqlSb.append(",");
+                if ("0".equals(isPrimaryKey) && myDbType.MySQL == myDbType) {
+                    //设置update语句
+                    updateSqlSb.append(targetFieldName);
+                    updateSqlSb.append(" =VALUES( ");
+                    updateSqlSb.append(targetFieldName);
+                    updateSqlSb.append(" ) ");
+                    updateSqlSb.append(",");
+                } else if (myDbType.PostgreSQL == myDbType) {
+                    if ("1".equals(isPrimaryKey)) {
+                        primaryKeySqlSb.append(" targetFieldName ");
+                        primaryKeySqlSb.append(",");
+                    } else {
+                        //设置update语句
+                        updateSqlSb.append(targetFieldName);
+                        updateSqlSb.append(" =EXCLUDED. ");
+                        updateSqlSb.append(targetFieldName);
+                        updateSqlSb.append(",");
+                    }
+                }
+            }
+        }
+
+        String primaryKeySqlStr = "";
+        if (org.apache.commons.lang3.StringUtils.isNotEmpty(primaryKeySqlSb)) {
+            primaryKeySqlStr = primaryKeySqlSb.substring(0, primaryKeySqlSb.length() - 1);
+        }
+        String updateSqlStr = "";
+        if (org.apache.commons.lang3.StringUtils.isNotEmpty(updateSqlSb)) {
+            updateSqlStr = updateSqlSb.substring(0, updateSqlSb.length() - 1);
+        }
+
+        StringBuilder sb = new StringBuilder();
+        if (myDbType.MySQL == myDbType) {
+            sb.append(" INSERT INTO ");
+            sb.append(tableName + " ");
+            sb.append(" VALUES( ");
+            sb.append(valuesSqlSb);
+            sb.append(" ) ");
+
+            sb.append(" ON DUPLICATE KEY UPDATE ");
+            sb.append(updateSqlStr);
+        } else if (myDbType.PostgreSQL == myDbType) {
+            sb.append(" INSERT INTO ");
+            sb.append(tableName + " ");
+            sb.append(" VALUES( ");
+            sb.append(valuesSqlSb);
+            sb.append(") ");
+
+            sb.append(" ON CONFLICT( ");
+            sb.append(primaryKeySqlStr);
+            sb.append(" ) do update ");
+            sb.append(updateSqlStr);
+        }
+        return sb.toString();
+    }
 }