Merge pull request #190 from datastax/issue/CDM-92

CDM-92: Fix NPE issue with target counter tables with null fields

Author: pravinbhat

RELEASE.md

@@ -1,5 +1,8 @@
 # Release Notes
-## [4.1.3] - 2023-07-06
+## [4.1.4] - 2023-08-16
+- In rare edge situations, counter tables with existing data in Target can have null values on target. This release will handle null values in Target counter table transparently.
+
+## [4.1.3] - 2023-08-14
 - Counter table columns will usually have zeros to begin with, but in rare edge situations, they can have null values. This release will handle null values in counter table transparently.
 
 ## [4.1.2] - 2023-07-06

SIT/smoke/04_counters/cdm.fixData.assert

@@ -1,7 +1,7 @@
-Read Record Count: 6
+Read Record Count: 7
 Mismatch Record Count: 2
 Corrected Mismatch Record Count: 2
 Missing Record Count: 1
 Corrected Missing Record Count: 0
-Valid Record Count: 3
+Valid Record Count: 4
 Skipped Record Count: 0

SIT/smoke/04_counters/cdm.fixForce.assert

@@ -1,7 +1,7 @@
-Read Record Count: 6
+Read Record Count: 7
 Mismatch Record Count: 0
 Corrected Mismatch Record Count: 0
 Missing Record Count: 1
 Corrected Missing Record Count: 1
-Valid Record Count: 5
+Valid Record Count: 6
 Skipped Record Count: 0

SIT/smoke/04_counters/cdm.migrateData.assert

@@ -1,4 +1,4 @@
-Read Record Count: 6
+Read Record Count: 7
 Skipped Record Count: 0
-Write Record Count: 6
+Write Record Count: 7
 Error Record Count: 0

SIT/smoke/04_counters/cdm.validateData.assert

@@ -1,7 +1,7 @@
-Read Record Count: 6
+Read Record Count: 7
 Mismatch Record Count: 0
 Corrected Mismatch Record Count: 0
 Missing Record Count: 0
 Corrected Missing Record Count: 0
-Valid Record Count: 6
+Valid Record Count: 7
 Skipped Record Count: 0

SIT/smoke/04_counters/expected.out

@@ -6,6 +6,7 @@
  record2 |    11 |    20
  record6 |   500 |  null
  record1 |     1 |     2
+ record7 |     1 |     2
  record4 | 10000 | 20001
 
-(6 rows)
+(7 rows)

SIT/smoke/04_counters/setup.cql

@@ -7,8 +7,11 @@ UPDATE origin.smoke_counters set col1=col1+1000, col2=col2+2000 where key='recor
 UPDATE origin.smoke_counters set col1=col1+10000, col2=col2+20000 where key='record4';
 UPDATE origin.smoke_counters set col1=col1+500, col2=col2+500 where key='record5';
 UPDATE origin.smoke_counters set col1=col1+500 where key='record6';
+UPDATE origin.smoke_counters set col1=col1+1, col2=col2+2 where key='record7';
 
 DROP TABLE IF EXISTS target.smoke_counters;
 CREATE TABLE target.smoke_counters(key text,col1 counter,col2 counter, PRIMARY KEY (key));
 
+UPDATE target.smoke_counters set col1=col1+1 where key='record7';
+
 SELECT * FROM origin.smoke_counters;

pom.xml

@@ -136,7 +136,7 @@
     <connection>scm:git:[email protected]:datastax/cassandra-data-migrator.git</connection>
     <developerConnection>scm:git:[email protected]:datastax/cassandra-data-migrator.git</developerConnection>
     <url>https://github.com/datastax/cassandra-data-migrator</url>
-    <tag>@{project.version}</tag>
+    <tag>@{version}</tag>
   </scm>
 
   <build>

src/main/java/com/datastax/cdm/cql/statement/TargetUpdateStatement.java

@@ -42,35 +42,38 @@ protected BoundStatement bind(Row originRow, Row targetRow, Integer ttl, Long wr
             boundStatement = boundStatement.set(currentBindIndex++, writeTime, Long.class);
         }
 
-        Object bindValue = null;
+        Object originValue, targetValue;
+        Object bindValueTarget = null;
         for (int targetIndex : columnIndexesToBind) {
             int originIndex = cqlTable.getCorrespondingIndex(targetIndex);
 
             try {
-                if(usingCounter && counterIndexes.contains(targetIndex)) {
-                    bindValue = cqlTable.getOtherCqlTable().getData(originIndex, originRow);
-                    if (null == bindValue) {
+                if (usingCounter && counterIndexes.contains(targetIndex)) {
+                    originValue = cqlTable.getOtherCqlTable().getData(originIndex, originRow);
+                    if (null == originValue) {
                         currentBindIndex++;
                         continue;
                     }
-                    bindValue = ((Long) bindValue - (null==targetRow ? 0 : (Long) cqlTable.getData(targetIndex, targetRow)));
+                    targetValue = (null == targetRow ? 0L : cqlTable.getData(targetIndex, targetRow));
+                    bindValueTarget = ((Long) originValue - (null == targetValue ? 0L : (Long) targetValue));
                 }
                 else if (targetIndex== explodeMapKeyIndex) {
-                    bindValue = explodeMapKey;
+                    bindValueTarget = explodeMapKey;
                 }
                 else if (targetIndex== explodeMapValueIndex) {
-                    bindValue = explodeMapValue;
+                    bindValueTarget = explodeMapValue;
                 } else {
                     if (originIndex < 0)
                         // we don't have data to bind for this column; continue to the next targetIndex
                         continue;
-                    bindValue = cqlTable.getOtherCqlTable().getAndConvertData(originIndex, originRow);
+                    bindValueTarget = cqlTable.getOtherCqlTable().getAndConvertData(originIndex, originRow);
                 }
 
-                boundStatement = boundStatement.set(currentBindIndex++, bindValue, cqlTable.getBindClass(targetIndex));
-            }
-            catch (Exception e) {
-                logger.error("Error trying to bind value:" + bindValue + " to column:" + targetColumnNames.get(targetIndex) + " of targetDataType:" + targetColumnTypes.get(targetIndex)+ "/" + cqlTable.getBindClass(targetIndex).getName() + " at column index:" + targetIndex);
+                boundStatement = boundStatement.set(currentBindIndex++, bindValueTarget, cqlTable.getBindClass(targetIndex));
+            } catch (Exception e) {
+                logger.error("Error trying to bind value:" + bindValueTarget + " to column:" +
+                        targetColumnNames.get(targetIndex) + " of targetDataType:" + targetColumnTypes.get(targetIndex) + "/"
+                        + cqlTable.getBindClass(targetIndex).getName() + " at column index:" + targetIndex);
                 throw e;
             }
         }