Merge pull request #16 from datastax/feature/ttl-writetime-independent

Feature/ttl writetime independent

Author: pravinbhat

pom.xml

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

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

@@ -36,20 +36,17 @@ protected AbstractJobSession(CqlSession sourceSession, CqlSession astraSession,
         sourceKeyspaceTable = sparkConf.get("spark.source.keyspaceTable");
         astraKeyspaceTable = sparkConf.get("spark.destination.keyspaceTable");
 
-        isPreserveTTLWritetime = Boolean.parseBoolean(sparkConf.get("spark.preserveTTLWriteTime", "false"));
-        if (isPreserveTTLWritetime) {
-            String ttlColsStr = sparkConf.get("spark.preserveTTLWriteTime.ttl.cols");
-            if (null != ttlColsStr && ttlColsStr.trim().length() > 0) {
-                for (String ttlCol : ttlColsStr.split(",")) {
-                    ttlCols.add(Integer.parseInt(ttlCol));
-                }
+        String ttlColsStr = sparkConf.get("spark.query.ttl.cols");
+        if (null != ttlColsStr && ttlColsStr.trim().length() > 0) {
+            for (String ttlCol : ttlColsStr.split(",")) {
+                ttlCols.add(Integer.parseInt(ttlCol));
             }
+        }
 
-            String writeTimestampColsStr = sparkConf.get("spark.preserveTTLWriteTime.writetime.cols");
-            if (null != writeTimestampColsStr && writeTimestampColsStr.trim().length() > 0) {
-                for (String writeTimeStampCol : writeTimestampColsStr.split(",")) {
-                    writeTimeStampCols.add(Integer.parseInt(writeTimeStampCol));
-                }
+        String writeTimestampColsStr = sparkConf.get("spark.query.writetime.cols");
+        if (null != writeTimestampColsStr && writeTimestampColsStr.trim().length() > 0) {
+            for (String writeTimeStampCol : writeTimestampColsStr.split(",")) {
+                writeTimeStampCols.add(Integer.parseInt(writeTimeStampCol));
             }
         }
 
@@ -82,26 +79,22 @@ protected AbstractJobSession(CqlSession sourceSession, CqlSession astraSession,
         logger.info("PARAM -- Destination Keyspace Table: " + astraKeyspaceTable);
         logger.info("PARAM -- ReadRateLimit: " + readLimiter.getRate());
         logger.info("PARAM -- WriteRateLimit: " + writeLimiter.getRate());
-        logger.info("PARAM -- WriteTimestampFilter: " + writeTimeStampFilter);
-        logger.info("PARAM -- WriteTimestampFilterCols: " + writeTimeStampCols);
-        logger.info("PARAM -- isPreserveTTLWritetime: " + isPreserveTTLWritetime);
-        logger.info("PARAM -- isPreserveTTLWritetime: " + isPreserveTTLWritetime);
         logger.info("PARAM -- TTLCols: " + ttlCols);
+        logger.info("PARAM -- WriteTimestampFilterCols: " + writeTimeStampCols);
+        logger.info("PARAM -- WriteTimestampFilter: " + writeTimeStampFilter);
 
         String selectCols = sparkConf.get("spark.query.source");
         String partionKey = sparkConf.get("spark.query.source.partitionKey");
         String sourceSelectCondition = sparkConf.get("spark.query.condition", "");
 
         final StringBuilder selectTTLWriteTimeCols = new StringBuilder();
-        if (isPreserveTTLWritetime) {
-            String[] allCols = selectCols.split(",");
-            ttlCols.forEach(col -> {
-                selectTTLWriteTimeCols.append(",ttl(" + allCols[col] + ")");
-            });
-            writeTimeStampCols.forEach(col -> {
-                selectTTLWriteTimeCols.append(",writetime(" + allCols[col] + ")");
-            });
-        }
+        String[] allCols = selectCols.split(",");
+        ttlCols.forEach(col -> {
+            selectTTLWriteTimeCols.append(",ttl(" + allCols[col] + ")");
+        });
+        writeTimeStampCols.forEach(col -> {
+            selectTTLWriteTimeCols.append(",writetime(" + allCols[col] + ")");
+        });
         String fullSelectQuery = "select " + selectCols + selectTTLWriteTimeCols.toString() + " from " + sourceKeyspaceTable + " where token(" + partionKey.trim()
                 + ") >= ? and token(" + partionKey.trim() + ") <= ?  " + sourceSelectCondition + " ALLOW FILTERING";
         sourceSelectStatement = sourceSession.prepare(fullSelectQuery);
@@ -110,7 +103,7 @@ protected AbstractJobSession(CqlSession sourceSession, CqlSession astraSession,
         selectColTypes = getTypes(sparkConf.get("spark.query.types"));
         String idCols = sparkConf.get("spark.query.destination.id", "");
         idColTypes = selectColTypes.subList(0, idCols.split(",").length);
-        
+
         String insertCols = sparkConf.get("spark.query.destination", "");
         if (null == insertCols || insertCols.trim().isEmpty()) {
             insertCols = selectCols;
@@ -147,11 +140,16 @@ protected AbstractJobSession(CqlSession sourceSession, CqlSession astraSession,
                 }
             }
 
-            if (isPreserveTTLWritetime) {
-                astraInsertStatement = astraSession.prepare("insert into " + astraKeyspaceTable + " (" + insertCols + ") VALUES (" + insertBinds + ") using TTL ? and TIMESTAMP ?");
-            } else {
-                astraInsertStatement = astraSession.prepare("insert into " + astraKeyspaceTable + " (" + insertCols + ") VALUES (" + insertBinds + ")");
+            String fullInsertQuery = "insert into " + astraKeyspaceTable + " (" + insertCols + ") VALUES (" + insertBinds + ")";
+            if (!ttlCols.isEmpty()) {
+                fullInsertQuery += " USING TTL ?";
+                if (!writeTimeStampCols.isEmpty()) {
+                    fullInsertQuery += " AND TIMESTAMP ?";
+                }
+            } else if (!writeTimeStampCols.isEmpty()) {
+                fullInsertQuery += " USING TIMESTAMP ?";
             }
+            astraInsertStatement = astraSession.prepare(fullInsertQuery);
         }
     }
 

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

@@ -32,7 +32,6 @@ public abstract class BaseJobSession {
     protected Integer batchSize = 1;
     protected Integer printStatsAfter = 100000;
 
-    protected Boolean isPreserveTTLWritetime = Boolean.FALSE;
     protected Boolean writeTimeStampFilter = Boolean.FALSE;
     protected Long minWriteTimeStampFilter = 0l;
     protected Long maxWriteTimeStampFilter = Long.MAX_VALUE;

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

@@ -173,9 +173,11 @@ public BoundStatement bindInsert(PreparedStatement insertStatement, Row sourceRo
                 }
             }
 
-            if (isPreserveTTLWritetime) {
+            if (!ttlCols.isEmpty()) {
                 boundInsertStatement = boundInsertStatement.set(index, getLargestTTL(sourceRow), Integer.class);
                 index++;
+            }
+            if (!writeTimeStampCols.isEmpty()) {
                 if (customWritetime > 0) {
                     boundInsertStatement = boundInsertStatement.set(index, customWritetime, Long.class);
                 } else {

src/resources/sparkConf.properties

@@ -28,15 +28,13 @@ spark.query.source.partitionKey                         partition-key
 spark.query.destination                                 partition-key,clustering-key,order-date,amount
 spark.query.destination.id                              partition-key,clustering-key
 spark.query.types                                       9,1,4,3
+spark.query.ttl.cols                                    2,3
+spark.query.writetime.cols                              2,3
 
 spark.counterTable                                      false
 spark.counterTable.cql
 spark.counterTable.cql.index                            0
 
-spark.preserveTTLWriteTime                              true
-spark.preserveTTLWriteTime.ttl.cols                     2,3
-spark.preserveTTLWriteTime.writetime.cols               2,3
-
 spark.source.writeTimeStampFilter                       false
 spark.source.minWriteTimeStampFilter                    0
 spark.source.maxWriteTimeStampFilter                    9223372036854775807