Merge pull request #14 from datastax/feature/custom_writetime

Feature/custom writetime

Author: faizalrub-datastax

README.md

@@ -70,3 +70,4 @@ spark.destination.autocorrect.mismatch                  true|false
 - SSL Support (including custom cipher algorithms)
 - Migrate from any Cassandra source ([Apache Cassandra](https://cassandra.apache.org)/[DataStax Enterprise (DSE)](https://www.datastax.com/products/datastax-enterprise)/[DataStax Astra DB](https://www.datastax.com/products/datastax-astra)) to any Cassandra target ([Apache Cassandra](https://cassandra.apache.org)/[DataStax Enterprise (DSE)](https://www.datastax.com/products/datastax-enterprise)/[DataStax Astra DB](https://www.datastax.com/products/datastax-astra))
 - Validate migration accuracy and performance using a smaller randomized data-set
+- Custom writetime

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

@@ -5,6 +5,7 @@
 import com.datastax.oss.driver.api.core.cql.PreparedStatement;
 import com.datastax.oss.driver.api.core.cql.Row;
 import com.datastax.oss.driver.shaded.guava.common.util.concurrent.RateLimiter;
+import org.apache.commons.lang.StringUtils;
 import org.apache.spark.SparkConf;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -70,6 +71,12 @@ protected AbstractJobSession(CqlSession sourceSession, CqlSession astraSession,
             maxWriteTimeStampFilter = Long.parseLong(maxWriteTimeStampFilterStr);
         }
 
+        String customWriteTimeStr =
+                sparkConf.get("spark.destination.custom.writeTime", "0");
+        if (null != customWriteTimeStr && customWriteTimeStr.trim().length() > 1 && StringUtils.isNumeric(customWriteTimeStr.trim())) {
+            customWritetime = Long.parseLong(customWriteTimeStr);
+        }
+
         logger.info("PARAM -- Write Batch Size: " + batchSize);
         logger.info("PARAM -- Source Keyspace Table: " + sourceKeyspaceTable);
         logger.info("PARAM -- Destination Keyspace Table: " + astraKeyspaceTable);

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

@@ -36,6 +36,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

@@ -176,7 +176,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

@@ -13,6 +13,7 @@ spark.destination.read.consistency.level                LOCAL_QUORUM
 spark.destination.keyspaceTable                         test.a2
 spark.destination.autocorrect.missing                   false
 spark.destination.autocorrect.mismatch                  false
+spark.destination.custom.writeTime 			            0
 
 spark.maxRetries                                        10
 spark.readRateLimit                                     20000
@@ -87,6 +88,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.destination.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,