Added custom writetime property. When set it will be used as writetime for target writes. This can be set to a value such as the historic migration start time to mitigate issues related to overwriting the real-time updates.

Author: faizalrub-datastax

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

@@ -68,6 +68,12 @@ protected AbstractJobSession(CqlSession sourceSession, CqlSession astraSession,
             maxWriteTimeStampFilter = Long.parseLong(maxWriteTimeStampFilterStr);
         }
 
+        String customWriteTimeStr =
+                sparkConf.get("spark.migrate.custom.writeTime", "0");
+        if (null != customWriteTimeStr && customWriteTimeStr.trim().length() > 1) {
+            customWritetime = Long.parseLong(customWriteTimeStr);
+        }
+
         logger.info(" DEFAULT -- Write Batch Size: " + batchSize);
         logger.info(" DEFAULT -- Source Keyspace Table: " + sourceKeyspaceTable);
         logger.info(" DEFAULT -- Destination Keyspace Table: " + astraKeyspaceTable);

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

@@ -40,6 +40,7 @@ public abstract  class BaseJobSession {
     protected Boolean writeTimeStampFilter = Boolean.FALSE;
     protected Long minWriteTimeStampFilter = 0l;
     protected Long maxWriteTimeStampFilter = Long.MAX_VALUE;
+    protected Long customWritetime = 0l;
 
     protected List<Integer> writeTimeStampCols = new ArrayList<Integer>();
     protected List<Integer> ttlCols = new ArrayList<Integer>();

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

@@ -211,7 +211,11 @@ public BoundStatement bindInsert(PreparedStatement insertStatement, Row sourceRo
             if (isPreserveTTLWritetime) {
                 boundInsertStatement = boundInsertStatement.set(index, getLargestTTL(sourceRow), Integer.class);
                 index++;
-                boundInsertStatement = boundInsertStatement.set(index, getLargestWriteTimeStamp(sourceRow), Long.class);
+                if (customWritetime > 0) {
+                    boundInsertStatement = boundInsertStatement.set(index, customWritetime, Long.class);
+                } else {
+                    boundInsertStatement = boundInsertStatement.set(index, getLargestWriteTimeStamp(sourceRow), Long.class);
+                }
             }
         }
 

src/resources/sparkConf.properties

@@ -42,6 +42,7 @@ spark.source.writeTimeStampFilter                       false
 spark.source.writeTimeStampFilter.cols                  4,5
 spark.source.minWriteTimeStampFilter                    0
 spark.source.maxWriteTimeStampFilter                    9223372036854775807
+spark.migrate.custom.writeTime 				            1664980600000000
 
 ########################## ONLY USE if SSL clientAuth is enabled on source Cassandra/DSE ###############################
 #spark.source.trustStore.path
@@ -90,6 +91,8 @@ spark.source.maxWriteTimeStampFilter                    9223372036854775807
 # "spark.source.writeTimeStampFilter.cols" - Comma separated column indexes from "spark.query.cols.select".
 #  Script will only use the largest value per row.
 #
+# "spark.migrate.custom.writeTime" - User specified write time. When set it will be used as writetime for target writes.
+#
 # Default value for "spark.source.maxWriteTimeStampFilter" is alway "9223372036854775807" (max long value)
 #
 # Properties "spark.query.cols.insert" and "spark.query.cols.insert.types" are required for "Migrate" job,