Permalink
Browse files

0002357: Need to be able to set the field terminator and record

terminator for the mssql bulk loader
  • Loading branch information...
chenson42 committed Jul 31, 2015
1 parent 0aa9952 commit 4045959fa81db512ebdeb0e6bad70752a1f6cfcd
@@ -20,8 +20,6 @@
*/
package org.jumpmind.symmetric.io;
import static org.apache.commons.lang.StringUtils.isNotBlank;
import java.io.OutputStream;
import java.sql.Connection;
import java.sql.SQLException;
@@ -70,10 +68,10 @@ public MsSqlBulkDatabaseWriter(IDatabasePlatform platform,
this.maxRowsBeforeFlush = maxRowsBeforeFlush;
this.stagingManager = stagingManager;
this.fireTriggers = fireTriggers;
if (isNotBlank(fieldTerminator)) {
if (fieldTerminator != null && fieldTerminator.length() > 0) {
this.fieldTerminator = fieldTerminator;
}
if (isNotBlank(rowTerminator)) {
if (rowTerminator != null && rowTerminator.length() > 0) {
this.rowTerminator = rowTerminator;
}
this.uncPath = uncPath;
@@ -205,11 +203,21 @@ protected void flush() {
String schemaSeparator = dbInfo.getSchemaSeparator();
JdbcSqlTransaction jdbcTransaction = (JdbcSqlTransaction) transaction;
Connection c = jdbcTransaction.getConnection();
String rowTerminatorString = "";
/*
* There seems to be a bug with the SQL server bulk insert when
* you have one row with binary data at the end using \n as the
* row terminator. It works when you leave the row terminator
* out of the bulk insert statement.
*/
if (!(rowTerminator.equals("\n") || rowTerminator.equals("\r\n"))) {
rowTerminatorString = ", ROWTERMINATOR='" + StringEscapeUtils.escapeJava(rowTerminator) + "'";
}
String sql = String.format("BULK INSERT " +
this.getTargetTable().getQualifiedTableName(quote, catalogSeparator, schemaSeparator) +
" FROM '" + filename) + "'" +
" WITH ( FIELDTERMINATOR='"+StringEscapeUtils.escapeJava(fieldTerminator)+"', KEEPIDENTITY" +
(fireTriggers ? ", FIRE_TRIGGERS" : "") + ", ROWTERMINATOR='"+StringEscapeUtils.escapeJava(rowTerminator)+"');";
(fireTriggers ? ", FIRE_TRIGGERS" : "") + rowTerminatorString +");";
Statement stmt = c.createStatement();
//TODO: clean this up, deal with errors, etc.?
@@ -26,15 +26,14 @@
import java.util.Map;
import java.util.Random;
import junit.framework.Assert;
import org.apache.commons.codec.binary.Base64;
import org.apache.commons.codec.binary.Hex;
import org.apache.commons.lang.ArrayUtils;
import org.jumpmind.db.platform.IDatabasePlatform;
import org.jumpmind.symmetric.io.AbstractWriterTest;
import org.jumpmind.symmetric.io.data.CsvData;
import org.jumpmind.symmetric.io.data.DataEventType;
import org.junit.Assert;
import org.junit.Test;
public abstract class AbstractBulkDatabaseWriterTest extends AbstractWriterTest {
@@ -45,14 +45,16 @@
public class MsSqlBulkDatabaseWriterTest extends AbstractBulkDatabaseWriterTest {
protected static IStagingManager stagingManager;
protected static final String uncPath = "\\\\192.168.42.121\\bulkloaddir";
@BeforeClass
public static void setup() throws Exception {
if (DbTestUtils.getEnvironmentSpecificProperties(DbTestUtils.ROOT).get(BasicDataSourcePropertyConstants.DB_POOL_DRIVER)
.equals("net.sourceforge.jtds.jdbc.Driver")) {
platform = DbTestUtils.createDatabasePlatform(DbTestUtils.ROOT);
platform.createDatabase(platform.readDatabaseFromXml("/testBulkWriter.xml", true), true, false);
stagingManager = new StagingManager("tmp");
stagingManager = new StagingManager("target/tmp");
}
}
@@ -70,7 +72,7 @@ protected boolean shouldTestRun(IDatabasePlatform platform) {
protected long writeData(List<CsvData> data) {
Table table = platform.getTableFromCache(getTestTable(), false);
return writeData(new MsSqlBulkDatabaseWriter(platform, stagingManager, new CommonsDbcpNativeJdbcExtractor(), 1000, false, null, "||", "\r\n"), new TableCsvData(table, data));
return writeData(new MsSqlBulkDatabaseWriter(platform, stagingManager, new CommonsDbcpNativeJdbcExtractor(), 1000, false, uncPath, null, null), new TableCsvData(table, data));
}
@Test
@@ -86,7 +88,7 @@ public void testInsertReorderColumns() throws Exception {
Column firstColumn = table.getColumn(0);
table.removeColumn(firstColumn);
table.addColumn(firstColumn);
writeData(new MsSqlBulkDatabaseWriter(platform, stagingManager, new CommonsDbcpNativeJdbcExtractor(), 1000, false, null, "||", "\r\n"),
writeData(new MsSqlBulkDatabaseWriter(platform, stagingManager, new CommonsDbcpNativeJdbcExtractor(), 1000, false, uncPath, null, null),
new TableCsvData(table, data));
values = (String[]) ArrayUtils.remove(values, values.length - 1);
values = (String[]) ArrayUtils.add(values, 0, id);

0 comments on commit 4045959

Please sign in to comment.