Implements workaround for C* bug (duplicates in list on insert/update with same timestamp) https://issues.apache.org/jira/browse/CASSANDRA-11368

Author: pravinbhat

pom.xml

@@ -8,7 +8,7 @@
 
   <properties>
     <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
-    <revision>3.4.1</revision>
+    <revision>3.4.2</revision>
     <scala.version>2.12.17</scala.version>
     <scala.main.version>2.12</scala.main.version>
     <spark.version>3.3.1</spark.version>

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

@@ -30,6 +30,8 @@ public class AbstractJobSession extends BaseJobSession {
     protected List<String> ttlWTCols;
     protected String tsReplaceValStr;
     protected long tsReplaceVal;
+    protected Long customWritetime = 0l;
+    protected Long incrWritetime = 0l;
 
     protected AbstractJobSession(CqlSession sourceSession, CqlSession astraSession, SparkConf sc) {
         this(sourceSession, astraSession, sc, false);
@@ -67,7 +69,6 @@ protected AbstractJobSession(CqlSession sourceSession, CqlSession astraSession,
         logger.info("PARAM -- Destination Table: {}", astraKeyspaceTable.split("\\.")[1]);
         logger.info("PARAM -- ReadRateLimit: {}", readLimiter.getRate());
         logger.info("PARAM -- WriteRateLimit: {}", writeLimiter.getRate());
-        logger.info("PARAM -- WriteTimestampFilter: {}", writeTimeStampFilter);
 
         tableInfo = TableInfo.getInstance(sourceSession, sourceKeyspaceTable.split("\\.")[0],
                 sourceKeyspaceTable.split("\\.")[1], Util.getSparkPropOrEmpty(sc, "spark.query.origin"));
@@ -95,13 +96,21 @@ protected AbstractJobSession(CqlSession sourceSession, CqlSession astraSession,
         }
 
         String customWriteTimeStr =
-                Util.getSparkPropOr(sc, "spark.target.custom.writeTime", "0");
-        if (null != customWriteTimeStr && customWriteTimeStr.trim().length() > 1 && StringUtils.isNumeric(customWriteTimeStr.trim())) {
+                Util.getSparkPropOr(sc, "spark.target.writeTime.fixedValue", "0");
+        if (null != customWriteTimeStr && customWriteTimeStr.trim().length() > 0 && StringUtils.isNumeric(customWriteTimeStr.trim())) {
             customWritetime = Long.parseLong(customWriteTimeStr);
         }
 
+        String incrWriteTimeStr =
+                Util.getSparkPropOr(sc, "spark.target.writeTime.incrementBy", "0");
+        if (null != incrWriteTimeStr && incrWriteTimeStr.trim().length() > 0 && StringUtils.isNumeric(incrWriteTimeStr.trim())) {
+            incrWritetime = Long.parseLong(incrWriteTimeStr);
+        }
+
         logger.info("PARAM -- TTL-WriteTime Columns: {}", ttlWTCols);
-        logger.info("PARAM -- WriteTimestampFilter: {}", writeTimeStampFilter);
+        logger.info("PARAM -- WriteTimes Filter: {}", writeTimeStampFilter);
+        logger.info("PARAM -- WriteTime Custom Value: {}", customWritetime);
+        logger.info("PARAM -- WriteTime increment Value: {}", incrWritetime);
         if (writeTimeStampFilter) {
             logger.info("PARAM -- minWriteTimeStampFilter: {} datetime is {}", minWriteTimeStampFilter,
                     Instant.ofEpochMilli(minWriteTimeStampFilter / 1000));
@@ -196,7 +205,7 @@ public BoundStatement bindInsert(PreparedStatement insertStatement, Row sourceRo
                 if (customWritetime > 0) {
                     boundInsertStatement = boundInsertStatement.set(index, customWritetime, Long.class);
                 } else {
-                    boundInsertStatement = boundInsertStatement.set(index, getLargestWriteTimeStamp(sourceRow), Long.class);
+                    boundInsertStatement = boundInsertStatement.set(index, getLargestWriteTimeStamp(sourceRow) + incrWritetime, Long.class);
                 }
             }
         }

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

@@ -40,7 +40,6 @@ public abstract class BaseJobSession {
     protected Boolean writeTimeStampFilter = Boolean.FALSE;
     protected Long minWriteTimeStampFilter = 0l;
     protected Long maxWriteTimeStampFilter = Long.MAX_VALUE;
-    protected Long customWritetime = 0l;
 
     protected Boolean isCounterTable = false;
 

src/resources/cdm.properties

@@ -91,7 +91,11 @@ spark.batchSize                                   10
 #spark.read.fetch.sizeInRows                       1000
 
 # ENABLE ONLY IF YOU WANT TO USE CUSTOM FIXED WRITETIME VALUE ON TARGET
-#spark.target.custom.writeTime                     0
+#spark.target.writeTime.fixedValue                 0
+
+# ENABLE ONLY IF YOU WANT TO INCREMENT SOURCE WRITETIME VALUE
+# DUPLICATES IN LIST FIELDS: USE THIS WORKAROUND FOR CASSANDRA BUG https://issues.apache.org/jira/browse/CASSANDRA-11368
+#spark.target.writeTime.incrementBy                0
 
 # ONLY USE when running in Guardrail mode to identify large fields
 #spark.guardrail.colSizeInKB                       1024