From d81d876f934ab827f05222cef096ded64c43a53b Mon Sep 17 00:00:00 2001
From: Eric Long <elong@jumpmind.com>
Date: Mon, 20 Sep 2021 14:29:07 -0400
Subject: [PATCH] 0005085: Data loader gets ArrayIndexOutOfBoundsException on
 batch

---
 .../io/data/writer/DefaultDatabaseWriter.java | 26 ++++++++++++++-----
 1 file changed, 19 insertions(+), 7 deletions(-)

diff --git a/symmetric-io/src/main/java/org/jumpmind/symmetric/io/data/writer/DefaultDatabaseWriter.java b/symmetric-io/src/main/java/org/jumpmind/symmetric/io/data/writer/DefaultDatabaseWriter.java
index ebee6c10ee..8ae250d638 100644
--- a/symmetric-io/src/main/java/org/jumpmind/symmetric/io/data/writer/DefaultDatabaseWriter.java
+++ b/symmetric-io/src/main/java/org/jumpmind/symmetric/io/data/writer/DefaultDatabaseWriter.java
@@ -450,21 +450,19 @@ protected LoadStatus update(CsvData data, boolean applyChangesOnly, boolean useC
             statistics.get(batch).startTimer(DataWriterStatisticConstants.LOADMILLIS);
             String[] rowData = getRowData(data, CsvData.ROW_DATA);
             String[] oldData = getRowData(data, CsvData.OLD_DATA);
-            ArrayList<String> changedColumnNameList = new ArrayList<String>();
-            ArrayList<String> changedColumnValueList = new ArrayList<String>();
-            ArrayList<Column> changedColumnsList = new ArrayList<Column>();
+            ArrayList<String> changedColumnValueList = new ArrayList<>();
+            ArrayList<Column> changedColumnsList = new ArrayList<>();
             for (int i = 0; i < targetTable.getColumnCount(); i++) {
                 Column column = targetTable.getColumn(i);
                 if (column != null) {
                     if (doesColumnNeedUpdated(i, column, data, rowData, oldData, applyChangesOnly)) {
-                        changedColumnNameList.add(column.getName());
                         changedColumnsList.add(column);
                         changedColumnValueList.add(rowData[i]);
                     }
                 }
             }
 
-            if (changedColumnNameList.size() > 0) {
+            if (changedColumnsList.size() > 0) {
                 Map<String, String> lookupDataMap = null;
                 Conflict conflict = writerSettings.pickConflict(this.targetTable, batch);
                 if (requireNewStatement(DmlType.UPDATE, data, applyChangesOnly,
@@ -1022,7 +1020,7 @@ protected boolean doesColumnNeedUpdated(int targetColumnIndex, Column column, Cs
                     || containsEmptyLobColumn;
             if (containsEmptyLobColumn) {
                 // indicate that we are considering the column to be changed
-                data.getChangedDataIndicators()[sourceTable.getColumnIndex(column.getName())] = true;
+                updateChangedDataIndicator(data, column, true);
             }
         } else {
             /*
@@ -1039,10 +1037,24 @@ protected boolean doesColumnNeedUpdated(int targetColumnIndex, Column column, Cs
              * A primary key change isn't indicated in the change data indicators when there is no old
              * data.  Need to update it manually in that case.
              */
-            data.getChangedDataIndicators()[sourceTable.getColumnIndex(column.getName())] = needsUpdated;
+            updateChangedDataIndicator(data, column, needsUpdated);
         }
         return needsUpdated;
     }
+    
+    protected void updateChangedDataIndicator(CsvData data, Column column, boolean needsUpdated) {
+        boolean[] changeIndicators = data.getChangedDataIndicators();
+        int index = sourceTable.getColumnIndex(column.getName());
+        if (index != -1 && index < changeIndicators.length) {
+            changeIndicators[index] = needsUpdated;
+        } else if (index == -1) {
+            log.warn("Unable to set change indicator because column {} not found on source table {}", column.getName(),
+                    sourceTable.getFullyQualifiedTableName());
+        } else {
+            log.warn("Unable to set change indicator because column {} is index {} on source table {}, but row data has only {} values",
+                    column.getName(), index, sourceTable.getFullyQualifiedTableName(), changeIndicators.length);
+        }
+    }
 
     protected void prepare() {
         getTransaction().prepare(this.currentDmlStatement.getSql());