fixes to data loader. reset sql transaction on rollback.

JumpMind · May 27, 2012 · 61d5304 · 61d5304
1 parent 0391de9
commit 61d5304
Show file tree

Hide file tree

Showing 8 changed files with 174 additions and 125 deletions.
diff --git a/...ymmetric-core/src/main/java/org/jumpmind/symmetric/service/impl/DataExtractorService.java b/...ymmetric-core/src/main/java/org/jumpmind/symmetric/service/impl/DataExtractorService.java
@@ -651,10 +651,9 @@ protected Table lookupAndOrderColumnsAccordingToTriggerHistory(String routerId,
                         "Could not find the table, %s, to extract",
                         Table.getFullyQualifiedTableName(catalogName, schemaName, tableName)));
             }
-            currentTable = currentTable.copy();
+            currentTable = currentTable.copyAndFilterColumns(triggerHistory.getParsedColumnNames(), triggerHistory.getParsedPkColumnNames(), true);
             currentTable.setCatalog(catalogName);
             currentTable.setSchema(schemaName);
-            currentTable.orderColumns(triggerHistory.getParsedColumnNames());
 
             Router router = triggerRouterService.getRouterById(routerId, false);
             if (router != null && setTargetTableName) {

diff --git a/...c/symmetric-core/src/main/java/org/jumpmind/symmetric/service/impl/DataLoaderService.java b/...c/symmetric-core/src/main/java/org/jumpmind/symmetric/service/impl/DataLoaderService.java
@@ -40,6 +40,7 @@
 import org.jumpmind.db.sql.ISqlRowMapper;
 import org.jumpmind.db.sql.ISqlTransaction;
 import org.jumpmind.db.sql.Row;
+import org.jumpmind.db.sql.SqlException;
 import org.jumpmind.db.sql.UniqueKeyException;
 import org.jumpmind.exception.IoException;
 import org.jumpmind.symmetric.ISymmetricEngine;
@@ -366,7 +367,7 @@ protected void logAndRethrow(Exception ex) throws IOException {
             }
             throw (IOException) ex;
         } else {
-            if (!(ex instanceof ConflictException)) {
+            if (!(ex instanceof ConflictException || ex instanceof SqlException)) {
                 log.error("Failed while parsing batch", ex);
             }
         }

diff --git a/...ymmetric-core/src/main/java/org/jumpmind/symmetric/service/impl/TriggerRouterService.java b/...ymmetric-core/src/main/java/org/jumpmind/symmetric/service/impl/TriggerRouterService.java
@@ -755,6 +755,7 @@ public void syncTriggers(StringBuilder sqlBuffer, boolean genAlways) {
 
                     // make sure channels are read from the database
                     configurationService.reloadChannels();
+
                     List<Trigger> triggersForCurrentNode = getTriggersForCurrentNode();
                     inactivateTriggers(triggersForCurrentNode, sqlBuffer);
                     updateOrCreateDatabaseTriggers(triggersForCurrentNode, sqlBuffer, genAlways);
@@ -899,7 +900,12 @@ protected void updateOrCreateDatabaseTriggers(List<Trigger> triggers, StringBuil
 
                 Set<Table> tables = getTablesForTrigger(trigger, triggers);
                 if (tables.size() > 0) {
-                    for (Table table : tables) {
+                    for (Table table : tables) {                        
+                        if (table.getPrimaryKeyColumnCount() == 0) {
+                            table = table.copy();
+                            table.makeAllColumnsPrimaryKeys();
+                        }
+
                         TriggerHistory latestHistoryBeforeRebuild = getNewestTriggerHistoryForTrigger(
                                 trigger.getTriggerId(),
                                 trigger.getSourceCatalogName(),

diff --git a/symmetric/symmetric-db/src/main/java/org/jumpmind/db/model/Table.java b/symmetric/symmetric-db/src/main/java/org/jumpmind/db/model/Table.java
@@ -98,7 +98,7 @@ public Table(String catalog, String schema, String tableName, String[] columnNam
         }
 
     }
-    
+
     public void removeAllColumns() {
         columns.clear();
     }
@@ -962,23 +962,6 @@ public Column[] getDistributionKeyColumns() {
         }
     }
 
-    public void reOrderColumns(Column[] targetOrder, boolean copyPrimaryKeys) {
-        ArrayList<Column> orderedColumns = new ArrayList<Column>(targetOrder.length);
-        for (int i = 0; i < targetOrder.length; i++) {
-            String name = targetOrder[i].getName();
-            for (Column column : columns) {
-                if (column.getName().equalsIgnoreCase(name)) {
-                    orderedColumns.add(column);
-                    if (copyPrimaryKeys) {
-                        column.setPrimaryKey(targetOrder[i].isPrimaryKey());
-                    }
-                    break;
-                }
-            }
-        }
-        columns = orderedColumns;
-    }
-
     public void orderColumns(String[] columnNames) {
         Column[] orderedColumns = orderColumns(columnNames, this);
         this.columns.clear();
@@ -1019,6 +1002,36 @@ public Table copy() {
         }
     }
 
+    public Table copyAndFilterColumns(String[] orderedColumnNames, String[] pkColumnNames,
+            boolean setPrimaryKeys) {
+        Table table = copy();
+        orderColumns(orderedColumnNames);
+
+        if (setPrimaryKeys) {
+            for (Column column : columns) {
+                column.setPrimaryKey(false);
+            }
+
+            if (pkColumnNames != null) {
+                for (Column column : columns) {
+                    for (String pkColumnName : pkColumnNames) {
+                        if (column.getName().equals(pkColumnName)) {
+                            column.setPrimaryKey(true);
+                        }
+                    }
+                }
+            }
+        }
+
+        return table;
+    }
+
+    public void makeAllColumnsPrimaryKeys() {
+        for (Column column : columns) {
+            column.setPrimaryKey(true);
+        }
+    }
+
     public String[] getColumnNames() {
         String[] columnNames = new String[columns.size()];
         int i = 0;
@@ -1037,7 +1050,7 @@ public String[] getPrimaryKeyColumnNames() {
         }
         return columnNames;
     }
-    
+
     public int calculateTableHashcode() {
         final int PRIME = 31;
         int result = 1;
@@ -1046,7 +1059,7 @@ public int calculateTableHashcode() {
         result = PRIME * result + calculateHashcodeForColumns(PRIME, getPrimaryKeyColumns());
         return result;
     }
-    
+
     private static int calculateHashcodeForColumns(final int PRIME, Column[] cols) {
         int result = 1;
         if (cols != null && cols.length > 0) {
@@ -1059,7 +1072,6 @@ private static int calculateHashcodeForColumns(final int PRIME, Column[] cols) {
         return result;
     }
 
-
     public static boolean areAllColumnsPrimaryKeys(Column[] columns) {
         boolean allPks = true;
         if (columns != null) {