Fix counter-table bug when run in Diff mode & simplify conter-table config

Author: pravinbhat

.idea/libraries/Maven__org_scala_lang_scala_library_2_11_8.xml

@@ -25,8 +25,6 @@ tar -xvzf <spark downloaded file name>
 ```
 ./spark-submit --properties-file sparkConf.properties /
 --master "local[*]" /
---conf spark.migrate.source.minPartition=-9223372036854775808 /
---conf spark.migrate.source.maxPartition=9223372036854775807 /
 --class datastax.astra.migrate.Migrate cassandra-data-migrator-1.x.jar &> logfile_name.txt
 ```
 
@@ -40,8 +38,6 @@ Note: Above command also generates a log file `logfile_name.txt` to avoid log ou
 ```
 ./spark-submit --properties-file sparkConf.properties /
 --master "local[*]" /
---conf spark.migrate.source.minPartition=-9223372036854775808 /
---conf spark.migrate.source.maxPartition=9223372036854775807 /
 --class datastax.astra.migrate.DiffData cassandra-data-migrator-1.x.jar &> logfile_name.txt
 ```
 
@@ -60,8 +56,8 @@ Note: Above command also generates a log file `logfile_name.txt` to avoid log ou
 - Enable/disable this feature using one or both of the below setting in the config file
 
 ```
-spark.migrate.destination.autocorrect.missing                   true|false
-spark.migrate.destination.autocorrect.mismatch                  true|false
+spark.destination.autocorrect.missing                   true|false
+spark.destination.autocorrect.mismatch                  true|false
 ```
 
 # Additional features

README.md

@@ -3,7 +3,7 @@
 
   <groupId>datastax.astra.migrate</groupId>
   <artifactId>cassandra-data-migrator</artifactId>
-  <version>1.1</version>
+  <version>1.2</version>
   <packaging>jar</packaging>
 
   <properties>

pom.xml

@@ -34,6 +34,7 @@ public abstract class AbstractJobSession {
 
     protected CqlSession sourceSession;
     protected CqlSession astraSession;
+    protected List<MigrateDataType> selectColTypes = new ArrayList<MigrateDataType>();
     protected List<MigrateDataType> idColTypes = new ArrayList<MigrateDataType>();
 
     protected Integer batchSize = 1;
@@ -47,7 +48,6 @@ public abstract class AbstractJobSession {
     protected List<Integer> writeTimeStampCols = new ArrayList<Integer>();
     protected List<Integer> ttlCols = new ArrayList<Integer>();
     protected Boolean isCounterTable;
-    protected Integer counterDeltaMaxIndex = 0;
 
     protected String sourceKeyspaceTable;
     protected String astraKeyspaceTable;
@@ -117,11 +117,8 @@ protected AbstractJobSession(CqlSession sourceSession, CqlSession astraSession,
 
         hasRandomPartitioner = Boolean.parseBoolean(sparkConf.get("spark.source.hasRandomPartitioner", "false"));
 
-        isCounterTable = Boolean.parseBoolean(sparkConf.get("spark.source.counterTable", "false"));
-
-        counterDeltaMaxIndex = Integer
-                .parseInt(sparkConf.get("spark.source.counterTable.update.max.counter.index", "0"));
-
+        isCounterTable = Boolean.parseBoolean(sparkConf.get("spark.counterTable", "false"));
+        selectColTypes = getTypes(sparkConf.get("spark.diff.select.types"));
         String partionKey = sparkConf.get("spark.query.cols.partitionKey");
         String idCols = sparkConf.get("spark.query.cols.id");
         idColTypes = getTypes(sparkConf.get("spark.query.cols.id.types"));

src/main/java/datastax/astra/migrate/AbstractJobSession.java

@@ -54,12 +54,12 @@ protected CopyJobSession(CqlSession sourceSession, CqlSession astraSession, Spar
         }
 
         if (isCounterTable) {
-            String updateSelectMappingStr = sparkConf.get("spark.source.counterTable.update.select.index", "0");
+            String updateSelectMappingStr = sparkConf.get("spark.counterTable.cql.index", "0");
             for (String updateSelectIndex : updateSelectMappingStr.split(",")) {
                 updateSelectMapping.add(Integer.parseInt(updateSelectIndex));
             }
 
-            String counterTableUpdate = sparkConf.get("spark.source.counterTable.update.cql");
+            String counterTableUpdate = sparkConf.get("spark.counterTable.cql");
             astraInsertStatement = astraSession.prepare(counterTableUpdate);
         } else {
             if (isPreserveTTLWritetime) {
@@ -181,8 +181,8 @@ public BoundStatement bindInsert(PreparedStatement insertStatement, Row sourceRo
             for (int index = 0; index < insertColTypes.size(); index++) {
                 MigrateDataType dataType = insertColTypes.get(index);
                 // compute the counter delta if reading from astra for the difference
-                if (astraRow != null && isCounterTable && index <= counterDeltaMaxIndex) {
-                    boundInsertStatement = boundInsertStatement.set(index, getCounterDelta(sourceRow.getLong(updateSelectMapping.get(index)), astraRow.getLong(updateSelectMapping.get(index))), Long.class);
+                if (astraRow != null && index < (selectColTypes.size() - idColTypes.size())) {
+                    boundInsertStatement = boundInsertStatement.set(index, (sourceRow.getLong(updateSelectMapping.get(index)) - astraRow.getLong(updateSelectMapping.get(index))), Long.class);
                 } else {
                     boundInsertStatement = boundInsertStatement.set(index, getData(dataType, updateSelectMapping.get(index), sourceRow), dataType.typeClass);
                 }
@@ -217,8 +217,4 @@ public BoundStatement bindInsert(PreparedStatement insertStatement, Row sourceRo
         return boundInsertStatement;
     }
 
-    public Long getCounterDelta(Long sourceRow, Long astraRow) {
-        return sourceRow - astraRow;
-    }
-
 }

src/main/java/datastax/astra/migrate/CopyJobSession.java

@@ -28,7 +28,6 @@ public class DiffJobSession extends CopyJobSession {
     private AtomicLong validCounter = new AtomicLong(0);
     private AtomicLong skippedCounter = new AtomicLong(0);
 
-    protected List<MigrateDataType> selectColTypes = new ArrayList<MigrateDataType>();
     protected Boolean autoCorrectMissing = false;
     protected Boolean autoCorrectMismatch = false;
 
@@ -47,7 +46,6 @@ public static DiffJobSession getInstance(CqlSession sourceSession, CqlSession as
     private DiffJobSession(CqlSession sourceSession, CqlSession astraSession, SparkConf sparkConf) {
         super(sourceSession, astraSession, sparkConf);
 
-        selectColTypes = getTypes(sparkConf.get("spark.diff.select.types"));
         autoCorrectMissing = Boolean.parseBoolean(sparkConf.get("spark.destination.autocorrect.missing", "false"));
         autoCorrectMismatch = Boolean.parseBoolean(sparkConf.get("spark.destination.autocorrect.mismatch", "false"));
     }
@@ -134,7 +132,11 @@ private void diff(Row sourceRow, Row astraRow) {
             logger.error("Data mismatch found -  Key: " + getKey(sourceRow) + " Data: " + diffData);
 
             if (autoCorrectMismatch) {
-                astraSession.execute(bindInsert(astraInsertStatement, sourceRow, null));
+                if (isCounterTable) {
+                    astraSession.execute(bindInsert(astraInsertStatement, sourceRow, astraRow));
+                } else {
+                    astraSession.execute(bindInsert(astraInsertStatement, sourceRow, null));
+                }
                 correctedMismatchCounter.incrementAndGet();
                 logger.error("Corrected mismatch data in Astra: " + getKey(sourceRow));
             }

src/main/java/datastax/astra/migrate/DiffJobSession.java

@@ -31,10 +31,9 @@ spark.query.cols.partitionKey                           partition-key
 spark.query.cols.insert                                 partition-key,clustering-key,order-date,amount
 spark.query.cols.insert.types                           9,1,4,3
 
-spark.source.counterTable                               false
-spark.source.counterTable.update.cql
-spark.source.counterTable.update.max.counter.index      0
-spark.source.counterTable.update.select.index           0
+spark.counterTable                                      false
+spark.counterTable.cql
+spark.counterTable.cql.index                            0
 
 spark.preserveTTLWriteTime                              true
 spark.source.ttl.cols                                   6,7