0004270: Oracle Bulk Loader: Allow specification of input file character

set
JumpMind · Jan 31, 2020 · c839fce · c839fce
1 parent c6c70b5
commit c839fce
Show file tree

Hide file tree

Showing 6 changed files with 27 additions and 4 deletions.
diff --git a/...ient/src/integrationTest/java/org/jumpmind/symmetric/io/OracleBulkDatabaseWriterTest.java b/...ient/src/integrationTest/java/org/jumpmind/symmetric/io/OracleBulkDatabaseWriterTest.java
@@ -83,7 +83,7 @@ protected long writeData(TableCsvData... datas) {
                 "silent=(header,discards) direct=false readsize=4096000 bindsize=4096000 rows=1000 discardmax=1 errors=0",
                 prop.get(BasicDataSourcePropertyConstants.DB_POOL_USER),
                 prop.get(BasicDataSourcePropertyConstants.DB_POOL_PASSWORD), 
-                prop.get(BasicDataSourcePropertyConstants.DB_POOL_URL), null, null), datas);
+                prop.get(BasicDataSourcePropertyConstants.DB_POOL_URL), null, null, null), datas);
     }
 
     @Override

diff --git a/symmetric-client/src/main/java/org/jumpmind/symmetric/ext/OracleBulkDataLoaderFactory.java b/symmetric-client/src/main/java/org/jumpmind/symmetric/ext/OracleBulkDataLoaderFactory.java
@@ -71,11 +71,13 @@ public IDataWriter getDataWriter(String sourceNodeId, ISymmetricDialect symmetri
 
         String sqlLoaderCommand = parmService.getString(ParameterConstants.DBDIALECT_ORACLE_BULK_LOAD_SQLLDR_CMD);
         String sqlLoaderOptions = parmService.getString(ParameterConstants.DBDIALECT_ORACLE_BULK_LOAD_SQLLDR_OPTIONS);
+        String sqlLoaderInfileCharset = parmService.getString(ParameterConstants.DBDIALECT_ORACLE_BULK_LOAD_SQLLDR_INFILE_CHARSET);
         String ezConnectString = parmService.getString(ParameterConstants.DBDIALECT_ORACLE_BULK_LOAD_EZCONNECT);
+
 
         return new OracleBulkDatabaseWriter(symmetricDialect.getPlatform(), symmetricDialect.getTargetPlatform(),
                 engine.getStagingManager(), engine.getTablePrefix(), sqlLoaderCommand, sqlLoaderOptions,
-                dbUser, dbPassword, dbUrl, ezConnectString,
+                dbUser, dbPassword, dbUrl, ezConnectString, sqlLoaderInfileCharset,
                 buildDatabaseWriterSettings(filters, errorHandlers, conflictSettings, resolvedData));
     }
 

diff --git a/symmetric-client/src/main/java/org/jumpmind/symmetric/io/OracleBulkDatabaseWriter.java b/symmetric-client/src/main/java/org/jumpmind/symmetric/io/OracleBulkDatabaseWriter.java
@@ -73,12 +73,15 @@ public class OracleBulkDatabaseWriter extends AbstractBulkDatabaseWriter {
     protected String dbUrl;
 
     protected String ezConnectString;
+
+    protected String sqlLoaderInfileCharset;
 
     protected int rows = 0;
 
     public OracleBulkDatabaseWriter(IDatabasePlatform symmetricPlatform, IDatabasePlatform targetPlatform,
             IStagingManager stagingManager, String tablePrefix, String sqlLoaderCommand, String sqlLoaderOptions,
-            String dbUser, String dbPassword, String dbUrl, String ezConnectString, DatabaseWriterSettings settings) {
+            String dbUser, String dbPassword, String dbUrl, String ezConnectString, String sqlLoaderInfileCharset,
+            DatabaseWriterSettings settings) {
         super(symmetricPlatform, targetPlatform, tablePrefix, settings);
         logger = LoggerFactory.getLogger(getClass());
         this.stagingManager = stagingManager;
@@ -87,6 +90,7 @@ public OracleBulkDatabaseWriter(IDatabasePlatform symmetricPlatform, IDatabasePl
         this.dbPassword = dbPassword;
         this.dbUrl = dbUrl;
         this.ezConnectString = StringUtils.defaultIfBlank(ezConnectString, getConnectString(dbUrl));
+        this.sqlLoaderInfileCharset = StringUtils.defaultIfBlank(sqlLoaderInfileCharset, null);
 
         this.sqlLoaderOptions = new ArrayList<String>();
         if (StringUtils.isNotBlank(sqlLoaderOptions)) {
@@ -137,6 +141,9 @@ protected void createStagingFile() {
         try {
             OutputStream out = controlResource.getOutputStream();
             out.write(("LOAD DATA\n").getBytes());
+            if(StringUtils.isNotEmpty(sqlLoaderInfileCharset)) {
+                out.write(("CHARACTERSET " + sqlLoaderInfileCharset + "\n").getBytes());
+            }
             out.write(getInfileControl().getBytes());
             out.write(("APPEND INTO TABLE " + targetTable.getQualifiedTableName("\"", ".", ".") + "\n").getBytes());
 

diff --git a/symmetric-client/src/main/java/org/jumpmind/symmetric/io/TiberoBulkDatabaseWriter.java b/symmetric-client/src/main/java/org/jumpmind/symmetric/io/TiberoBulkDatabaseWriter.java
@@ -34,7 +34,7 @@ public TiberoBulkDatabaseWriter(IDatabasePlatform symmetricPlatform, IDatabasePl
             IStagingManager stagingManager, String tablePrefix, String tbLoaderCommand, String tbLoaderOptions,
             String dbUser, String dbPassword, String dbUrl, String dbName, DatabaseWriterSettings settings) {
         super(symmetricPlatform, targetPlatform, stagingManager, tablePrefix, tbLoaderCommand, tbLoaderOptions,
-                dbUser, dbPassword, dbUrl, dbName, settings);
+                dbUser, dbPassword, dbUrl, dbName, null, settings);
         logger = LoggerFactory.getLogger(getClass());
     }
 

diff --git a/symmetric-core/src/main/java/org/jumpmind/symmetric/common/ParameterConstants.java b/symmetric-core/src/main/java/org/jumpmind/symmetric/common/ParameterConstants.java
@@ -255,6 +255,7 @@ private ParameterConstants() {
     public final static String DBDIALECT_ORACLE_SEQUENCE_NOORDER_NEXTVALUE_DB_URLS = "oracle.sequence.noorder.nextvalue.db.urls";
     public final static String DBDIALECT_ORACLE_BULK_LOAD_SQLLDR_CMD = "oracle.bulk.load.sqlldr.cmd";
     public final static String DBDIALECT_ORACLE_BULK_LOAD_SQLLDR_OPTIONS = "oracle.bulk.load.sqlldr.options";
+    public final static String DBDIALECT_ORACLE_BULK_LOAD_SQLLDR_INFILE_CHARSET = "oracle.bulk.load.sqlldr.infile.charset";
     public final static String DBDIALECT_ORACLE_BULK_LOAD_EZCONNECT = "oracle.bulk.load.ezconnect";
     public final static String DBDIALECT_ORACLE_LOAD_QUERY_HINT_PARALLEL_COUNT = "oracle.load.query.hint.parallel.count";
 

diff --git a/symmetric-core/src/main/resources/symmetric-default.properties b/symmetric-core/src/main/resources/symmetric-default.properties
@@ -1833,6 +1833,19 @@ oracle.bulk.load.sqlldr.options=silent=(header,discards) direct=false readsize=4
 # DatabaseOverridable: false
 oracle.bulk.load.ezconnect=
 
+# For bulk loading with SQL*Loader, specify input file character set
+# when needed to support characters other than US-ASCII characters.
+# For example:
+# oracle.bulk.load.sqlldr.infile.charset=UTF8 LENGTH SEMANTICS CHAR
+# This will generate the following string in the control file after
+# the LOAD DATA line:
+# CHARACTERSET UTF8 LENGTH SEMANTICS CHAR
+# If blank, this line will not show up. 
+#
+# Tags: other
+# DatabaseOverridable: false
+oracle.bulk.load.sqlldr.infile.charset=
+
 # For initial load extracting data to specify the number of parallel processes to use while selecting data from a table
 #
 # Tags: other