353 lines
15 KiB
Plaintext
353 lines
15 KiB
Plaintext
<%@ jet
|
|
imports="
|
|
org.talend.designer.codegen.config.CodeGeneratorArgument
|
|
org.talend.core.model.process.INode
|
|
org.talend.core.model.process.ElementParameterParser
|
|
org.talend.core.model.metadata.IMetadataTable
|
|
org.talend.core.model.process.IConnection
|
|
org.talend.core.model.metadata.MappingTypeRetriever
|
|
org.talend.core.model.metadata.MetadataTalendType
|
|
java.util.List
|
|
java.util.ArrayList
|
|
java.util.Map
|
|
java.util.HashMap
|
|
"
|
|
skeleton="../templates/db_output_bulk.skeleton"
|
|
%>
|
|
<%@ include file="@{org.talend.designer.components.localprovider}/components/templates/Log4j/DBLogUtil.javajet"%>
|
|
<%
|
|
CodeGeneratorArgument codeGenArgument = (CodeGeneratorArgument) argument;
|
|
INode node = (INode)codeGenArgument.getArgument();
|
|
|
|
String cid = node.getUniqueName();
|
|
|
|
dbLog = new DBLogUtil(node);
|
|
|
|
List<Map<String, String>> addCols =
|
|
(List<Map<String,String>>)ElementParameterParser.getObjectValue(
|
|
node,"__ADD_COLS__" );
|
|
|
|
boolean useFieldOptions = ("true").equals(ElementParameterParser.getValue(node, "__USE_FIELD_OPTIONS__"));
|
|
|
|
List<Map<String, String>> fieldOptions = (List<Map<String,String>>)ElementParameterParser.getObjectValue(node, "__FIELD_OPTIONS__");
|
|
|
|
String dbname= ElementParameterParser.getValue(node, "__DBNAME__");
|
|
|
|
String dbhost = ElementParameterParser.getValue(node, "__HOST__");
|
|
|
|
String dbport = ElementParameterParser.getValue(node, "__PORT__");
|
|
|
|
String dbschema = ElementParameterParser.getValue(node, "__SCHEMA_DB__");
|
|
|
|
String dbuser= ElementParameterParser.getValue(node, "__USER__");
|
|
|
|
String tableName = ElementParameterParser.getValue(node,"__TABLE__");
|
|
|
|
String dbmsId = ElementParameterParser.getValue(node,"__MAPPING__");
|
|
|
|
String dataAction = ElementParameterParser.getValue(node,"__DATA_ACTION__");
|
|
|
|
String tableAction = ElementParameterParser.getValue(node,"__TABLE_ACTION__");
|
|
|
|
String commitEvery = ElementParameterParser.getValue(node, "__COMMIT_EVERY__");
|
|
|
|
String useExistingConn = ElementParameterParser.getValue(node,"__USE_EXISTING_CONNECTION__");
|
|
|
|
boolean isEnableDebug = ("true").equals(ElementParameterParser.getValue(node,"__ENABLE_DEBUG_MODE__"));
|
|
boolean isParallelize ="true".equalsIgnoreCase(ElementParameterParser.getValue(node, "__PARALLELIZE__"));
|
|
|
|
boolean useBatch = "true".equals(ElementParameterParser.getValue(node, "__USE_BATCH__"));
|
|
String batchSize = ElementParameterParser.getValue(node, "__BATCH_SIZE__");
|
|
|
|
String rejectConnName = null;
|
|
List<? extends IConnection> rejectConns = node.getOutgoingConnections("REJECT");
|
|
if(rejectConns != null && rejectConns.size() > 0) {
|
|
IConnection rejectConn = rejectConns.get(0);
|
|
rejectConnName = rejectConn.getName();
|
|
}
|
|
|
|
useBatch = (rejectConnName == null)
|
|
&& (
|
|
("INSERT").equals(dataAction) || ("UPDATE").equals(dataAction) || ("DELETE").equals(dataAction)
|
|
)
|
|
&& useBatch;
|
|
%>
|
|
|
|
<%
|
|
|
|
getManager(dbmsId, cid, node);
|
|
|
|
List<IMetadataColumn> columnList = getColumnList(node);
|
|
List<Column> stmtStructure = null;
|
|
if(columnList != null && columnList.size() > 0) {
|
|
stmtStructure = getManager(dbmsId, cid).createColumnList(columnList, useFieldOptions, fieldOptions, addCols);
|
|
}
|
|
%>
|
|
|
|
String dbschema_<%=cid%> = null;
|
|
<%
|
|
if(("true").equals(useExistingConn)) {
|
|
%>
|
|
dbschema_<%=cid%> = (String)globalMap.get("schema_" + "<%=ElementParameterParser.getValue(node,"__CONNECTION__")%>");
|
|
<%
|
|
} else {
|
|
%>
|
|
dbschema_<%=cid%> = <%=dbschema%>;
|
|
<%
|
|
}
|
|
%>
|
|
|
|
String tableName_<%=cid%> = null;
|
|
if(dbschema_<%=cid%> == null || dbschema_<%=cid%>.trim().length() == 0) {
|
|
tableName_<%=cid%> = <%=tableName%>;
|
|
} else {
|
|
tableName_<%=cid%> = dbschema_<%=cid%> + "\".\"" + <%=tableName%>;
|
|
}
|
|
|
|
<%
|
|
if(("UPDATE").equals(dataAction) || ("INSERT_OR_UPDATE").equals(dataAction) || ("UPDATE_OR_INSERT").equals(dataAction)) {
|
|
int updateKeyCount = 0;
|
|
if(stmtStructure != null) {
|
|
for(Column column : stmtStructure) {
|
|
if(column.isUpdateKey()) {
|
|
updateKeyCount++;
|
|
}
|
|
}
|
|
%>
|
|
int updateKeyCount_<%=cid%> = <%=updateKeyCount%>;
|
|
if(updateKeyCount_<%=cid%> < 1) {
|
|
throw new RuntimeException("For update, Schema must have a key");
|
|
}
|
|
<%
|
|
}
|
|
} else if(("DELETE").equals(dataAction)) {
|
|
int deleteKeyCount = 0;
|
|
if(stmtStructure != null) {
|
|
for(Column column : stmtStructure) {
|
|
if(column.isDeleteKey()) {
|
|
deleteKeyCount++;
|
|
}
|
|
}
|
|
%>
|
|
int deleteKeyCount_<%=cid%> = <%=deleteKeyCount%>;
|
|
if(deleteKeyCount_<%=cid%> < 1) {
|
|
throw new RuntimeException("For delete, Schema must have a key");
|
|
}
|
|
<%
|
|
}
|
|
}
|
|
%>
|
|
int nb_line_<%=cid%> = 0;
|
|
int nb_line_update_<%=cid%> = 0;
|
|
int nb_line_inserted_<%=cid%> = 0;
|
|
int nb_line_deleted_<%=cid%> = 0;
|
|
int nb_line_rejected_<%=cid%> = 0;
|
|
|
|
int deletedCount_<%=cid%>=0;
|
|
int updatedCount_<%=cid%>=0;
|
|
int insertedCount_<%=cid%>=0;
|
|
int rejectedCount_<%=cid%>=0;
|
|
|
|
boolean whetherReject_<%=cid%> = false;
|
|
|
|
<%
|
|
if (useBatch) {
|
|
%>
|
|
int batchSize_<%=cid%> = <%=batchSize%>;
|
|
int batchSizeCounter_<%=cid%>=0;
|
|
int tmp_batchUpdateCount_<%=cid%> = 0;
|
|
<%
|
|
}
|
|
%>
|
|
|
|
java.sql.Connection conn_<%=cid%> = null;
|
|
<%
|
|
if(("true").equals(useExistingConn)) {
|
|
String connection = ElementParameterParser.getValue(node,"__CONNECTION__");
|
|
String conn = "conn_" + connection;
|
|
%>
|
|
conn_<%=cid%> = (java.sql.Connection)globalMap.get("<%=conn%>");
|
|
<%dbLog.conn().useExistConn("conn_"+cid+".getMetaData().getURL()", "conn_"+cid+".getMetaData().getUserName()");%>
|
|
<%
|
|
} else {
|
|
%>
|
|
<%dbLog.conn().logJDBCDriver(dbLog.str("org.postgresql.Driver"));%>
|
|
java.lang.Class.forName("org.postgresql.Driver");
|
|
String url_<%=cid %> = "jdbc:postgresql://"+<%=dbhost%>+":"+<%=dbport%>+"/"+<%=dbname%>;
|
|
String dbUser_<%=cid %> = <%=dbuser%>;
|
|
<%
|
|
String passwordFieldName = "__PASS__";
|
|
%>
|
|
|
|
<%@ include file="@{org.talend.designer.components.localprovider}/components/templates/password.javajet"%>
|
|
|
|
String dbPwd_<%=cid %> = decryptedPassword_<%=cid%>;
|
|
|
|
<%dbLog.conn().connTry(dbLog.var("url"), dbLog.var("dbUser"));%>
|
|
conn_<%=cid%> = java.sql.DriverManager.getConnection(url_<%=cid %>,dbUser_<%=cid%>,dbPwd_<%=cid%>);
|
|
<%dbLog.conn().connDone(dbLog.var("url"));%>
|
|
resourceMap.put("conn_<%=cid%>", conn_<%=cid%>);
|
|
<%
|
|
}
|
|
if(!("true").equals(useExistingConn)) {
|
|
if(!("").equals(commitEvery) && !("0").equals(commitEvery)) {
|
|
%>
|
|
conn_<%=cid%>.setAutoCommit(false);
|
|
int commitEvery_<%=cid%> = <%=commitEvery%>;
|
|
int commitCounter_<%=cid%> = 0;
|
|
<%
|
|
}
|
|
}
|
|
dbLog.commit().logAutoCommit("conn_"+cid+".getAutoCommit()");
|
|
%>
|
|
|
|
<%
|
|
if(columnList != null && columnList.size()>0) {
|
|
if(!isParallelize && !("NONE").equals(tableAction)) {
|
|
Manager manager = getManager(dbmsId, cid);
|
|
if(("DROP_CREATE").equals(tableAction)) {
|
|
%>
|
|
java.sql.Statement stmtDrop_<%=cid%> = conn_<%=cid%>.createStatement();
|
|
<%dbLog.table().dropTry(dbLog.var("tableName"));%>
|
|
stmtDrop_<%=cid%>.execute("<%=manager.getDropTableSQL()%>");
|
|
<%dbLog.table().dropDone(dbLog.var("tableName"));%>
|
|
stmtDrop_<%=cid%>.close();
|
|
|
|
java.sql.Statement stmtCreate_<%=cid%> = conn_<%=cid%>.createStatement();
|
|
<%dbLog.table().createTry(dbLog.var("tableName"));%>
|
|
stmtCreate_<%=cid%>.execute("<%=manager.getCreateTableSQL(stmtStructure)%>)");
|
|
<%dbLog.table().createDone(dbLog.var("tableName"));%>
|
|
stmtCreate_<%=cid%>.close();
|
|
<%
|
|
} else if(("CREATE").equals(tableAction)) {
|
|
%>
|
|
java.sql.Statement stmtCreate_<%=cid%> = conn_<%=cid%>.createStatement();
|
|
<%dbLog.table().createTry(dbLog.var("tableName"));%>
|
|
stmtCreate_<%=cid%>.execute("<%=manager.getCreateTableSQL(stmtStructure)%>)");
|
|
<%dbLog.table().createDone(dbLog.var("tableName"));%>
|
|
stmtCreate_<%=cid%>.close();
|
|
<%
|
|
} else if(("CREATE_IF_NOT_EXISTS").equals(tableAction) || ("DROP_IF_EXISTS_AND_CREATE").equals(tableAction)) {
|
|
%>
|
|
java.sql.DatabaseMetaData dbMetaData_<%=cid%> = conn_<%=cid%>.getMetaData();
|
|
java.sql.ResultSet rsTable_<%=cid%> = dbMetaData_<%=cid%>.getTables(null, null, null, new String[]{"TABLE"});
|
|
boolean whetherExist_<%=cid%> = false;
|
|
String defaultSchema_<%=cid%> = "public";
|
|
if(dbschema_<%=cid%> == null || dbschema_<%=cid%>.trim().length() == 0) {
|
|
java.sql.Statement stmtSchema_<%=cid%> = conn_<%=cid%>.createStatement();
|
|
java.sql.ResultSet rsSchema_<%=cid%> = stmtSchema_<%=cid%>.executeQuery("select current_schema() ");
|
|
while(rsSchema_<%=cid%>.next()){
|
|
defaultSchema_<%=cid%> = rsSchema_<%=cid%>.getString("current_schema");
|
|
}
|
|
rsSchema_<%=cid%>.close();
|
|
stmtSchema_<%=cid%>.close();
|
|
}
|
|
while(rsTable_<%=cid%>.next()) {
|
|
String table_<%=cid%> = rsTable_<%=cid%>.getString("TABLE_NAME");
|
|
String schema_<%=cid%> = rsTable_<%=cid%>.getString("TABLE_SCHEM");
|
|
if(table_<%=cid%>.equals(<%=tableName%>)
|
|
&& (schema_<%=cid%>.equals(dbschema_<%=cid%>) || ((dbschema_<%=cid%> == null || dbschema_<%=cid%>.trim().length() ==0) && defaultSchema_<%=cid%>.equals(schema_<%=cid%>)))) {
|
|
whetherExist_<%=cid%> = true;
|
|
break;
|
|
}
|
|
}
|
|
rsTable_<%=cid%>.close();
|
|
<%
|
|
if(("CREATE_IF_NOT_EXISTS").equals(tableAction)) {
|
|
%>
|
|
if(!whetherExist_<%=cid%>) {
|
|
java.sql.Statement stmtCreate_<%=cid%> = conn_<%=cid%>.createStatement();
|
|
<%dbLog.table().createTry(dbLog.var("tableName"));%>
|
|
stmtCreate_<%=cid%>.execute("<%=manager.getCreateTableSQL(stmtStructure)%>)");
|
|
<%dbLog.table().createDone(dbLog.var("tableName"));%>
|
|
stmtCreate_<%=cid%>.close();
|
|
}
|
|
<%
|
|
} else {
|
|
%>
|
|
if(whetherExist_<%=cid%>) {
|
|
java.sql.Statement stmtDrop_<%=cid%> = conn_<%=cid%>.createStatement();
|
|
<%dbLog.table().dropTry(dbLog.var("tableName"));%>
|
|
stmtDrop_<%=cid%>.execute("<%=manager.getDropTableSQL()%>");
|
|
<%dbLog.table().dropDone(dbLog.var("tableName"));%>
|
|
stmtDrop_<%=cid%>.close();
|
|
}
|
|
java.sql.Statement stmtCreate_<%=cid%> = conn_<%=cid%>.createStatement();
|
|
<%dbLog.table().createTry(dbLog.var("tableName"));%>
|
|
stmtCreate_<%=cid%>.execute("<%=manager.getCreateTableSQL(stmtStructure)%>)");
|
|
<%dbLog.table().createDone(dbLog.var("tableName"));%>
|
|
stmtCreate_<%=cid%>.close();
|
|
<%
|
|
}
|
|
} else if(("CLEAR").equals(tableAction)) {
|
|
%>
|
|
java.sql.Statement stmtClear_<%=cid%> = conn_<%=cid%>.createStatement();
|
|
<%dbLog.table().clearTry(dbLog.var("tableName"));%>
|
|
deletedCount_<%=cid%> = deletedCount_<%=cid%> + stmtClear_<%=cid%>.executeUpdate("<%=manager.getDeleteTableSQL()%>");
|
|
<%dbLog.table().clearDone(dbLog.var("tableName"));%>
|
|
stmtClear_<%=cid%>.close();
|
|
<%
|
|
} else if(("TRUNCATE").equals(tableAction)) {
|
|
%>
|
|
java.sql.Statement stmtTruncCount_<%=cid%> = conn_<%=cid%>.createStatement();
|
|
java.sql.ResultSet rsTruncCount_<%=cid%> = stmtTruncCount_<%=cid%>.executeQuery("<%=manager.getSelectionSQL()%>");
|
|
java.sql.Statement stmtTrunc_<%=cid%> = conn_<%=cid%>.createStatement();
|
|
<%dbLog.table().truncateTry(dbLog.var("tableName"));%>
|
|
stmtTrunc_<%=cid%>.executeUpdate("<%=manager.getTruncateTableSQL()%>");
|
|
<%dbLog.table().truncateDone(dbLog.var("tableName"));%>
|
|
while(rsTruncCount_<%=cid%>.next()) {
|
|
deletedCount_<%=cid%> += rsTruncCount_<%=cid%>.getInt(1);
|
|
}
|
|
rsTruncCount_<%=cid%>.close();
|
|
stmtTruncCount_<%=cid%>.close();
|
|
stmtTrunc_<%=cid%>.close();
|
|
<%
|
|
}
|
|
}
|
|
Map<String, StringBuilder> actionSQLMap = getManager(dbmsId, cid).createProcessSQL(stmtStructure);
|
|
StringBuilder insertColName = actionSQLMap.get(INSERT_COLUMN_NAME);
|
|
StringBuilder insertValueStmt = actionSQLMap.get(INSERT_VALUE_STMT);
|
|
StringBuilder updateSetStmt = actionSQLMap.get(UPDATE_SET_STMT);
|
|
StringBuilder updateWhereStmt = actionSQLMap.get(UPDATE_WHERE_STMT);
|
|
StringBuilder deleteWhereStmt = actionSQLMap.get(DELETE_WHERE_STMT);
|
|
|
|
if(("INSERT").equals(dataAction)) {
|
|
%>
|
|
String insert_<%=cid%> = "INSERT INTO \"" + tableName_<%=cid%> + "\" (<%=insertColName.toString()%>) VALUES (<%=insertValueStmt.toString()%>)";
|
|
java.sql.PreparedStatement pstmt_<%=cid %> = conn_<%=cid%>.prepareStatement(insert_<%=cid%>);
|
|
<%
|
|
} else if (("UPDATE").equals(dataAction)) {
|
|
%>
|
|
String update_<%=cid%> = "UPDATE \"" + tableName_<%=cid%> + "\" SET <%=updateSetStmt.toString()%> WHERE <%=updateWhereStmt.toString()%>";
|
|
java.sql.PreparedStatement pstmt_<%=cid %> = conn_<%=cid%>.prepareStatement(update_<%=cid%>);
|
|
<%
|
|
} else if (("INSERT_OR_UPDATE").equals(dataAction)) {
|
|
%>
|
|
java.sql.PreparedStatement pstmt_<%=cid %> = conn_<%=cid%>.prepareStatement("SELECT COUNT(1) FROM \"" + tableName_<%=cid%> + "\" WHERE <%=updateWhereStmt.toString()%>");
|
|
String insert_<%=cid%> = "INSERT INTO \"" + tableName_<%=cid%> + "\" (<%=insertColName.toString()%>) VALUES (<%=insertValueStmt.toString()%>)";
|
|
java.sql.PreparedStatement pstmtInsert_<%=cid %> = conn_<%=cid%>.prepareStatement(insert_<%=cid%>);
|
|
String update_<%=cid%> = "UPDATE \"" + tableName_<%=cid%> + "\" SET <%=updateSetStmt.toString()%> WHERE <%=updateWhereStmt.toString()%>";
|
|
java.sql.PreparedStatement pstmtUpdate_<%=cid %> = conn_<%=cid%>.prepareStatement(update_<%=cid%>);
|
|
<%
|
|
} else if (("UPDATE_OR_INSERT").equals(dataAction)) {
|
|
%>
|
|
String update_<%=cid%> = "UPDATE \"" + tableName_<%=cid%> + "\" SET <%=updateSetStmt.toString()%> WHERE <%=updateWhereStmt.toString()%>";
|
|
java.sql.PreparedStatement pstmtUpdate_<%=cid %> = conn_<%=cid%>.prepareStatement(update_<%=cid%>);
|
|
String insert_<%=cid%> = "INSERT INTO \"" + tableName_<%=cid%> + "\" (<%=insertColName.toString()%>) VALUES (<%=insertValueStmt.toString()%>)";
|
|
java.sql.PreparedStatement pstmtInsert_<%=cid %> = conn_<%=cid%>.prepareStatement(insert_<%=cid%>);
|
|
<%
|
|
} else if (("DELETE").equals(dataAction)) {
|
|
%>
|
|
String delete_<%=cid%> = "DELETE FROM \"" + tableName_<%=cid%> + "\" WHERE <%=deleteWhereStmt.toString()%>";
|
|
java.sql.PreparedStatement pstmt_<%=cid %> = conn_<%=cid%>.prepareStatement(delete_<%=cid%>);
|
|
<%
|
|
}
|
|
if(isEnableDebug) {
|
|
%>
|
|
StringBuffer query_<%=cid%> = null;
|
|
<%@ include file="../templates/DB/Output/splitSQLForAllDBInBegin.javajet" %>
|
|
<%
|
|
}
|
|
}
|
|
%>
|