0001942: PostGIS initial load fails unless you set initial.load.conca…

…t.csv.in.sql.enabled=true
JumpMind · Sep 22, 2014 · 2be8cb8 · 2be8cb8
1 parent 48ee4e9
commit 2be8cb8
Show file tree

Hide file tree

Showing 2 changed files with 25 additions and 18 deletions.
diff --git a/symmetric-core/src/main/java/org/jumpmind/symmetric/db/AbstractTriggerTemplate.java b/symmetric-core/src/main/java/org/jumpmind/symmetric/db/AbstractTriggerTemplate.java
@@ -124,19 +124,18 @@ public String createInitalLoadSql(Node node, TriggerRouter triggerRouter, Table
 
         String sql = null;
 
+        String tableAlias = symmetricDialect.getInitialLoadTableAlias();
         if (parameterService.is(
                 ParameterConstants.INITIAL_LOAD_CONCAT_CSV_IN_SQL_ENABLED)) {
             sql = sqlTemplates.get(INITIAL_LOAD_SQL_TEMPLATE);
-            String columnsText = buildColumnsString(symmetricDialect.getInitialLoadTableAlias(),
-                    symmetricDialect.getInitialLoadTableAlias(), "", columns, DataEventType.INSERT,
+            String columnsText = buildColumnsString(tableAlias,
+                    tableAlias, "", columns, DataEventType.INSERT,
                     false, channel, triggerRouter.getTrigger()).columnString;
             if (isNotBlank(textColumnExpression)) {
                 columnsText = textColumnExpression.replace("$(columnName)", columnsText);
             }            
             sql = FormatUtils.replace("columns", columnsText, sql);
         } else {            
-            boolean dateTimeAsString = parameterService.is(
-                    ParameterConstants.DATA_LOADER_TREAT_DATETIME_AS_VARCHAR);
             sql = "select $(columns) from $(schemaName)$(tableName) t where $(whereClause)";
             StringBuilder columnList = new StringBuilder();
             for (int i = 0; i < columns.length; i++) {
@@ -149,16 +148,14 @@ public String createInitalLoadSql(Node node, TriggerRouter triggerRouter, Table
                             columnList.append(",");
                         }
 
-                        String columnExpression = SymmetricUtils.quote(symmetricDialect,
-                                column.getName()); 
-
-                        if (dateTimeAsString && TypeMap.isDateTimeType(column.getMappedTypeCode())) {
-                            columnExpression = castDatetimeColumnToString(column.getName());
-                        } else if (isNotBlank(textColumnExpression) && TypeMap.isTextType(column.getMappedTypeCode())) {
-                            columnExpression = textColumnExpression.replace("$(columnName)", columnExpression) + " as " + column.getName();
+                        ColumnString columnString = fillOutColumnTemplate(tableAlias, tableAlias,
+                                "", column, DataEventType.INSERT, false, channel, triggerRouter.getTrigger());
+                        String columnExpression = columnString.columnString;
+                        if (isNotBlank(textColumnExpression) && TypeMap.isTextType(column.getMappedTypeCode())) {
+                            columnExpression = textColumnExpression.replace("$(columnName)", columnExpression);
                         }
 
-                        columnList.append(columnExpression);
+                        columnList.append(columnExpression).append(" as ").append(column.getName());
 
                     }
                 }

diff --git a/symmetric-core/src/main/java/org/jumpmind/symmetric/service/impl/DataExtractorService.java b/symmetric-core/src/main/java/org/jumpmind/symmetric/service/impl/DataExtractorService.java
@@ -37,6 +37,7 @@
 
 import org.apache.commons.lang.StringUtils;
 import org.jumpmind.db.io.DatabaseXmlUtil;
+import org.jumpmind.db.model.Column;
 import org.jumpmind.db.model.Database;
 import org.jumpmind.db.model.Table;
 import org.jumpmind.db.platform.DatabaseNamesConstants;
@@ -1594,17 +1595,26 @@ protected void startNewCursor(final TriggerHistory triggerHistory,
                     configurationService.getChannel(triggerRouter.getTrigger().getChannelId()),
                     overrideSelectSql);
             this.cursor = sqlTemplate.queryForCursor(initialLoadSql, new ISqlRowMapper<Data>() {
-                public Data mapRow(Row rs) {
+                public Data mapRow(Row row) {
 
                     String csvRow = null;
                     if (symmetricDialect.getParameterService().is(
                             ParameterConstants.INITIAL_LOAD_CONCAT_CSV_IN_SQL_ENABLED)) {
-                        csvRow = rs.stringValue();
+                        csvRow = row.stringValue();
                     } else {
-                        String[] rowData = platform.getStringValues(
-                                symmetricDialect.getBinaryEncoding(), sourceTable.getColumns(), rs,
-                                false);
-                        csvRow = CsvUtils.escapeCsvData(rowData, '\0', '"');
+                        StringBuilder concatedRow = new StringBuilder();
+                        Column[] columns = sourceTable.getColumns();
+                        int index = 0;
+                        for (Column column : columns) {
+                            if (index > 0) {
+                                concatedRow.append(",");
+                            }
+                            String value = row.getString(column.getName());
+                            concatedRow.append(value == null ? "" : value);
+                            index++;
+                        }
+                        csvRow = concatedRow.toString();
+
                     }
                     int commaCount = StringUtils.countMatches(csvRow, ",");
                     if (expectedCommaCount <= commaCount) {