Skip to content
Permalink
Browse files

0003830: Tibero bulk loader using tbLoader

  • Loading branch information...
elong
elong committed Dec 17, 2018
1 parent a1e670d commit d5d0f586cdac036d68f94f8484ce93e981c5d30c
@@ -29,7 +29,7 @@
import org.jumpmind.symmetric.ISymmetricEngine;
import org.jumpmind.symmetric.common.ParameterConstants;
import org.jumpmind.symmetric.db.ISymmetricDialect;
import org.jumpmind.symmetric.io.OracleBulkDatabaseWriter;
import org.jumpmind.symmetric.io.TiberoBulkDatabaseWriter;
import org.jumpmind.symmetric.io.data.IDataWriter;
import org.jumpmind.symmetric.io.data.writer.Conflict;
import org.jumpmind.symmetric.io.data.writer.IDatabaseWriterErrorHandler;
@@ -71,11 +71,11 @@ public IDataWriter getDataWriter(String sourceNodeId, ISymmetricDialect symmetri

String tbLoaderCommand = parmService.getString(ParameterConstants.DBDIALECT_TIBERO_BULK_LOAD_TBLOADER_CMD);
String tbLoaderOptions = parmService.getString(ParameterConstants.DBDIALECT_TIBERO_BULK_LOAD_TBLOADER_OPTIONS);
String ezConnectString = parmService.getString(ParameterConstants.DBDIALECT_TIBERO_BULK_LOAD_EZCONNECT);
String dbName = parmService.getString(ParameterConstants.DBDIALECT_TIBERO_BULK_LOAD_DBNAME);

return new OracleBulkDatabaseWriter(symmetricDialect.getPlatform(), symmetricDialect.getTargetPlatform(),
return new TiberoBulkDatabaseWriter(symmetricDialect.getPlatform(), symmetricDialect.getTargetPlatform(),
engine.getStagingManager(), engine.getTablePrefix(), tbLoaderCommand, tbLoaderOptions,
dbUser, dbPassword, dbUrl, ezConnectString,
dbUser, dbPassword, dbUrl, dbName,
buildDatabaseWriterSettings(filters, errorHandlers, conflictSettings, resolvedData));
}

@@ -50,7 +50,7 @@

protected final static String LINE_TERMINATOR = "|>";

protected final Logger logger = LoggerFactory.getLogger(getClass());
protected Logger logger;

protected IStagingManager stagingManager;

@@ -80,12 +80,13 @@ public OracleBulkDatabaseWriter(IDatabasePlatform symmetricPlatform, IDatabasePl
IStagingManager stagingManager, String tablePrefix, String sqlLoaderCommand, String sqlLoaderOptions,
String dbUser, String dbPassword, String dbUrl, String ezConnectString, DatabaseWriterSettings settings) {
super(symmetricPlatform, targetPlatform, tablePrefix, settings);
logger = LoggerFactory.getLogger(getClass());
this.stagingManager = stagingManager;
this.sqlLoaderCommand = sqlLoaderCommand;
this.dbUser = dbUser;
this.dbPassword = dbPassword;
this.dbUrl = dbUrl;
this.ezConnectString = StringUtils.defaultIfBlank(ezConnectString, getEzConnectString(dbUrl));
this.ezConnectString = StringUtils.defaultIfBlank(ezConnectString, getConnectString(dbUrl));

this.sqlLoaderOptions = new ArrayList<String>();
if (StringUtils.isNotBlank(sqlLoaderOptions)) {
@@ -307,7 +308,7 @@ protected void cleanup(boolean inError) {
}
}

protected String getEzConnectString(String dbUrl) {
protected String getConnectString(String dbUrl) {
String ezConnect = null;
int index = dbUrl.indexOf("@//");
if (index != -1) {
@@ -26,28 +26,26 @@
import org.jumpmind.db.platform.IDatabasePlatform;
import org.jumpmind.symmetric.io.data.writer.DatabaseWriterSettings;
import org.jumpmind.symmetric.io.stage.IStagingManager;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class TiberoBulkDatabaseWriter extends OracleBulkDatabaseWriter {

protected final Logger logger = LoggerFactory.getLogger(getClass());

public TiberoBulkDatabaseWriter(IDatabasePlatform symmetricPlatform, IDatabasePlatform targetPlatform,
IStagingManager stagingManager, String tablePrefix, String tbLoaderCommand, String tbLoaderOptions,
String dbUser, String dbPassword, String dbUrl, String ezConnectString, DatabaseWriterSettings settings) {
String dbUser, String dbPassword, String dbUrl, String dbName, DatabaseWriterSettings settings) {
super(symmetricPlatform, targetPlatform, stagingManager, tablePrefix, tbLoaderCommand, tbLoaderOptions,
dbUser, dbPassword, dbUrl, ezConnectString, settings);
dbUser, dbPassword, dbUrl, dbName, settings);
logger = LoggerFactory.getLogger(getClass());
}

@Override
protected void init() {
if (StringUtils.isBlank(this.sqlLoaderCommand)) {
String oracleHome = System.getenv("TB_HOME");
if (StringUtils.isNotBlank(oracleHome)) {
this.sqlLoaderCommand = oracleHome + File.separator + "bin" + File.separator + "tbldr";
String tiberoHome = System.getenv("TB_HOME");
if (StringUtils.isNotBlank(tiberoHome)) {
this.sqlLoaderCommand = tiberoHome + File.separator + "client" + File.separator + "bin" + File.separator + "tbloader";
} else {
this.sqlLoaderCommand = "tbldr";
this.sqlLoaderCommand = "tbloader";
}
}
}
@@ -59,11 +57,22 @@ protected String getInfileControl() {

@Override
protected String getLineTerminatedByControl() {
return "LINES TERMINATED BY '" + LINE_TERMINATOR + "'";
return "LINES TERMINATED BY '" + LINE_TERMINATOR + "'\n";
}

@Override
protected String getLoaderName() {
return "TBLoader";
}

@Override
protected String getConnectString(String dbUrl) {
String connectStr = "";
int index = dbUrl.lastIndexOf(":");
if (index != -1) {
connectStr = "@" + dbUrl.substring(index + 1);
}
return connectStr;
}

}
@@ -241,7 +241,7 @@ private ParameterConstants() {
public final static String DBDIALECT_TIBERO_USE_HINTS = "tibero.use.hints";
public final static String DBDIALECT_TIBERO_BULK_LOAD_TBLOADER_CMD = "tibero.bulk.load.tbloader.cmd";
public final static String DBDIALECT_TIBERO_BULK_LOAD_TBLOADER_OPTIONS = "tibero.bulk.load.tbloader.options";
public final static String DBDIALECT_TIBERO_BULK_LOAD_EZCONNECT = "tibero.bulk.load.ezconnect";
public final static String DBDIALECT_TIBERO_BULK_LOAD_DBNAME = "tibero.bulk.load.dbname";

public final static String DBDIALECT_ORACLE_TRANSACTION_VIEW_CLOCK_SYNC_THRESHOLD_MS = "oracle.transaction.view.clock.sync.threshold.ms";

@@ -1624,14 +1624,14 @@ tibero.bulk.load.tbloader.cmd=
# Tags: other
# Type: boolean
# DatabaseOverridable: false
tibero.bulk.load.tbloader.options=direct=N parallel=1 disable_idx=N readsize=2097152 bindsize=2097152 rows=2000 errors=0
tibero.bulk.load.tbloader.options=direct=N dpl_parallel=1 disable_idx=N readsize=2097152 bindsize=2097152 rows=2000 errors=0

# For bulk loading with SQL*Loader, specify how to connect to the database with an ezconnect name.
# If blank, the connection is determined using the db.url parameter.
# For bulk loading with tbLoader, specify the database name.
# If blank, the database name is determined using the db.url parameter.
#
# Tags: other
# DatabaseOverridable: false
tibero.bulk.load.ezconnect=
tibero.bulk.load.dname=

# Use to map the version string a zseries jdbc driver returns to the 'zseries' dialect
# Tags: other

0 comments on commit d5d0f58

Please sign in to comment.
You can’t perform that action at this time.