Merge pull request #118 from datastax/issue/CDM-27

mieslep · web-flow · commit a3cbb77712b1 · 2023-04-06T16:30:47.000+01:00
CDM-27 implementing TTL/Writettime tests and SIT, and fixing problems
diff --git a/SIT/common.sh b/SIT/common.sh
@@ -1,10 +1,10 @@
-DOCKER_CASS=cdm-sit-cass
-DOCKER_CDM=cdm-sit-cdm
-CASS_USERNAME=cassandra
-CASS_PASSWORD=cassandra
-KEYSPACES="source target"
-CDM_DIRECTORY=/local
-CDM_JARFILE=cassandra-data-migrator.jar
+export DOCKER_CASS=cdm-sit-cass
+export DOCKER_CDM=cdm-sit-cdm
+export CASS_USERNAME=cassandra
+export CASS_PASSWORD=cassandra
+export KEYSPACES="source target"
+export CDM_DIRECTORY=/local
+export CDM_JARFILE=cassandra-data-migrator.jar
 
 _info() {
   echo "INFO $*"
diff --git a/SIT/smoke/03_ttl_writetime/alternateCheckResults.sh b/SIT/smoke/03_ttl_writetime/alternateCheckResults.sh
@@ -0,0 +1,4 @@
+#!/bin/bash
+# We need TTL to go down by at least 1 second, e.g. make 60000 to 59999
+sleep 1
+docker exec ${DOCKER_CASS} /bin/bash -c "cqlsh -u $CASS_USERNAME -p $CASS_PASSWORD -f $testDir/expected.cql | tr -d ' ' | cut -c1-46"
diff --git a/SIT/smoke/03_ttl_writetime/breakData.cql b/SIT/smoke/03_ttl_writetime/breakData.cql
@@ -0,0 +1,27 @@
+
+DELETE FROM target.smoke_ttl_writetime USING TIMESTAMP 1087383780000000 WHERE key='record4'; -- 11:00:03
+INSERT INTO source.smoke_ttl_writetime(key, t_col1) VALUES ('record4','AAAA') USING TTL 4000 AND TIMESTAMP 1087384200000000; -- 11:10:00
+INSERT INTO source.smoke_ttl_writetime(key,tw_col2) VALUES ('record4','BBBB') USING TTL 4000 AND TIMESTAMP 1087384200000000; -- 11:10:00
+INSERT INTO source.smoke_ttl_writetime(key, w_col3) VALUES ('record4','CCCC') USING TTL 4000 AND TIMESTAMP 1087384200000000; -- 11:10:00
+INSERT INTO source.smoke_ttl_writetime(key,   col4) VALUES ('record4','DDDD') USING TTL 4000 AND TIMESTAMP 1087384200000000; -- 11:10:00
+
+
+--  t_col1 is TTL only, so the TTL value should be picked by the diff, but not the writetime
+INSERT INTO source.smoke_ttl_writetime(key, t_col1)  VALUES ('record5','A---') USING TTL 700000 AND TIMESTAMP 1087384200000000; -- 11:10:00
+INSERT INTO source.smoke_ttl_writetime(key, tw_col2) VALUES ('record5','B---') USING TTL 700000 AND TIMESTAMP 1087384200000000; -- 11:10:00
+
+--  t_col1 is TTL only. Its timestamp will update correctly, but the change will be written with previous timestamps
+--    from tw_col2 and w_col3, which means the change will be lost
+INSERT INTO source.smoke_ttl_writetime(key, t_col1)  VALUES ('record8','A---') USING TTL 700000 AND TIMESTAMP 1087384200000000; -- 11:10:00
+
+
+--  w_col3 is Writetime only, so the writetime value should be picked by the diff, but not the diff
+INSERT INTO source.smoke_ttl_writetime(key, w_col3)  VALUES ('record6','C---') USING TTL 700000 AND TIMESTAMP 1087384200000000; -- 11:10:00
+
+-- tw_col2 is TTL and Writetime, so both changes should be propatated by the diff
+INSERT INTO source.smoke_ttl_writetime(key, tw_col2) VALUES ('record7','B---') USING TTL 700000 AND TIMESTAMP 1087384200000000; -- 11:10:00
+
+
+
+SELECT * FROM target.smoke_ttl_writetime;
+
diff --git a/SIT/smoke/03_ttl_writetime/cdm.fixData.assert b/SIT/smoke/03_ttl_writetime/cdm.fixData.assert
@@ -0,0 +1,7 @@
+Read Record Count: 8
+Mismatch Record Count: 4
+Corrected Mismatch Record Count: 4
+Missing Record Count: 1
+Corrected Missing Record Count: 1
+Valid Record Count: 3
+Skipped Record Count: 0
diff --git a/SIT/smoke/03_ttl_writetime/cdm.migrateData.assert b/SIT/smoke/03_ttl_writetime/cdm.migrateData.assert
@@ -0,0 +1,4 @@
+Read Record Count: 8
+Skipped Record Count: 0
+Write Record Count: 8
+Error Record Count: 0
diff --git a/SIT/smoke/03_ttl_writetime/cdm.txt b/SIT/smoke/03_ttl_writetime/cdm.txt
@@ -0,0 +1,3 @@
+migrateData datastax.astra.migrate.Migrate migrate.properties
+validateData datastax.astra.migrate.DiffData migrate.properties
+fixData datastax.astra.migrate.DiffData fix.properties
diff --git a/SIT/smoke/03_ttl_writetime/cdm.validateData.assert b/SIT/smoke/03_ttl_writetime/cdm.validateData.assert
@@ -0,0 +1,7 @@
+Read Record Count: 8
+Mismatch Record Count: 0
+Corrected Mismatch Record Count: 0
+Missing Record Count: 0
+Corrected Missing Record Count: 0
+Valid Record Count: 8
+Skipped Record Count: 0
diff --git a/SIT/smoke/03_ttl_writetime/execute.sh b/SIT/smoke/03_ttl_writetime/execute.sh
@@ -0,0 +1,15 @@
+#!/bin/bash -e
+
+workingDir="$1"
+cd "$workingDir"
+
+/local/cdm.sh -f cdm.txt -s migrateData -d "$workingDir" > cdm.migrateData.out 2>cdm.migrateData.err
+/local/cdm-assert.sh -f cdm.migrateData.out -a cdm.migrateData.assert -d "$workingDir"
+
+/local/cdm.sh -f cdm.txt -s validateData -d "$workingDir" > cdm.validateData.out 2>cdm.validateData.err
+/local/cdm-assert.sh -f cdm.validateData.out -a cdm.validateData.assert -d "$workingDir"
+
+cqlsh -u $CASS_USERNAME -p $CASS_PASSWORD $CASS_CLUSTER -f $workingDir/breakData.cql > $workingDir/breakData.out 2> $workingDir/breakData.err
+
+/local/cdm.sh -f cdm.txt -s fixData -d "$workingDir" > cdm.fixData.out 2>cdm.fixData.err
+/local/cdm-assert.sh -f cdm.fixData.out -a cdm.fixData.assert -d "$workingDir"
diff --git a/SIT/smoke/03_ttl_writetime/expected.cql b/SIT/smoke/03_ttl_writetime/expected.cql
@@ -0,0 +1 @@
+SELECT key,t_col1,tw_col2,w_col3,col4, writetime(col4) as wt_col4, ttl(col4) as ttl_col4 FROM target.smoke_ttl_writetime;
diff --git a/SIT/smoke/03_ttl_writetime/expected.out b/SIT/smoke/03_ttl_writetime/expected.out
@@ -0,0 +1,13 @@
+
+key|t_col1|tw_col2|w_col3|col4|wt_col4|ttl_col
+---------+--------+---------+--------+------+-
+record3|AAAA|BBBB|CCCC|DDDD|1087383720000000|5
+record5|A---|B---|CCCC|DDDD|1087384200000000|6
+record2|AAAA|BBBB|CCCC|DDDD|1087383720000000|5
+record6|AAAA|BBBB|C---|DDDD|1087384200000000|5
+record1|AAAA|BBBB|CCCC|DDDD|1087383720000000|5
+record8|AAAA|BBBB|CCCC|DDDD|1087383720000000|6
+record7|AAAA|B---|CCCC|DDDD|1087384200000000|6
+record4|AAAA|BBBB|CCCC|DDDD|1087384200000000|3
+
+(8rows)
diff --git a/SIT/smoke/03_ttl_writetime/fix.properties b/SIT/smoke/03_ttl_writetime/fix.properties
@@ -0,0 +1,24 @@
+spark.origin.host                                 cdm-sit-cass
+spark.origin.port                                 9042
+spark.origin.username                             cassandra
+spark.origin.password                             cassandra
+spark.origin.keyspaceTable                        source.smoke_ttl_writetime
+
+spark.target.host                                 cdm-sit-cass
+spark.target.port                                 9042
+spark.target.username                             cassandra
+spark.target.password                             cassandra
+spark.target.keyspaceTable                        target.smoke_ttl_writetime
+
+spark.numSplits                                   1
+
+spark.query.origin                                key,t_col1,tw_col2,w_col3,col4
+spark.query.types                                 0,0,0,0,0
+spark.query.origin.partitionKey                   key
+spark.query.target.id                             key
+
+spark.query.ttl.cols                              1,2
+spark.query.writetime.cols                        2,3
+
+spark.target.autocorrect.missing                  true
+spark.target.autocorrect.mismatch                 true
diff --git a/SIT/smoke/03_ttl_writetime/migrate.properties b/SIT/smoke/03_ttl_writetime/migrate.properties
@@ -0,0 +1,21 @@
+spark.origin.host                                 cdm-sit-cass
+spark.origin.port                                 9042
+spark.origin.username                             cassandra
+spark.origin.password                             cassandra
+spark.origin.keyspaceTable                        source.smoke_ttl_writetime
+
+spark.target.host                                 cdm-sit-cass
+spark.target.port                                 9042
+spark.target.username                             cassandra
+spark.target.password                             cassandra
+spark.target.keyspaceTable                        target.smoke_ttl_writetime
+
+spark.numSplits                                   1
+
+spark.query.origin                                key,t_col1,tw_col2,w_col3,col4
+spark.query.types                                 0,0,0,0,0
+spark.query.origin.partitionKey                   key
+spark.query.target.id                             key
+
+spark.query.ttl.cols                              1,2
+spark.query.writetime.cols                        2,3
diff --git a/SIT/smoke/03_ttl_writetime/setup.cql b/SIT/smoke/03_ttl_writetime/setup.cql
@@ -0,0 +1,71 @@
+DROP TABLE IF EXISTS source.smoke_ttl_writetime;
+CREATE TABLE source.smoke_ttl_writetime(key text, t_col1 text,tw_col2 text,w_col3 text,col4 text, PRIMARY KEY (key));
+
+-- record1 : expect TTL≈50000 from A and writetime :02 1087383720000000 from C
+INSERT INTO source.smoke_ttl_writetime(key, t_col1) VALUES ('record1','AAAA') USING TTL  60000 AND TIMESTAMP 1087383600000000; -- 11:00:00
+INSERT INTO source.smoke_ttl_writetime(key,tw_col2) VALUES ('record1','BBBB') USING TTL   5000 AND TIMESTAMP 1087383660000000; -- 11:00:01
+INSERT INTO source.smoke_ttl_writetime(key, w_col3) VALUES ('record1','CCCC') USING TTL    400 AND TIMESTAMP 1087383720000000; -- 11:00:02
+INSERT INTO source.smoke_ttl_writetime(key,   col4) VALUES ('record1','DDDD') USING TTL 700000 AND TIMESTAMP 1087383780000000; -- 11:00:03
+
+-- record2 : expect TTL≈50000 from B and writetime 02: 1087383720000000 from B
+INSERT INTO source.smoke_ttl_writetime(key, t_col1) VALUES ('record2','AAAA') USING TTL   5000 AND TIMESTAMP 1087383600000000; -- 11:00:00
+INSERT INTO source.smoke_ttl_writetime(key,tw_col2) VALUES ('record2','BBBB') USING TTL  60000 AND TIMESTAMP 1087383720000000; -- 11:00:02
+INSERT INTO source.smoke_ttl_writetime(key, w_col3) VALUES ('record2','CCCC') USING TTL    400 AND TIMESTAMP 1087383660000000; -- 11:00:01
+INSERT INTO source.smoke_ttl_writetime(key,   col4) VALUES ('record2','DDDD') USING TTL 700000 AND TIMESTAMP 1087383780000000; -- 11:00:03
+
+-- record3 : expect TTL≈50000 from B and writetime 02: 1087383720000000 from C
+INSERT INTO source.smoke_ttl_writetime(key, t_col1) VALUES ('record3','AAAA') USING TTL   5000 AND TIMESTAMP 1087383600000000; -- 11:00:00
+INSERT INTO source.smoke_ttl_writetime(key,tw_col2) VALUES ('record3','BBBB') USING TTL  60000 AND TIMESTAMP 1087383660000000; -- 11:00:01
+INSERT INTO source.smoke_ttl_writetime(key, w_col3) VALUES ('record3','CCCC') USING TTL    400 AND TIMESTAMP 1087383720000000; -- 11:00:02
+INSERT INTO source.smoke_ttl_writetime(key,   col4) VALUES ('record3','DDDD') USING TTL 700000 AND TIMESTAMP 1087383780000000; -- 11:00:03
+
+-- record4: this is to test missing record. we will delete target.record4 with a later timestamp 11:00:03 
+--          and then create a new source.record4 with an even later timestamp than that 11:10:00.
+INSERT INTO source.smoke_ttl_writetime(key, t_col1) VALUES ('record4','AAAA') USING TTL 4000 AND TIMESTAMP 1087383600000000; -- 11:00:00
+INSERT INTO source.smoke_ttl_writetime(key,tw_col2) VALUES ('record4','BBBB') USING TTL 4000 AND TIMESTAMP 1087383600000000; -- 11:00:00
+INSERT INTO source.smoke_ttl_writetime(key, w_col3) VALUES ('record4','CCCC') USING TTL 4000 AND TIMESTAMP 1087383600000000; -- 11:00:00
+INSERT INTO source.smoke_ttl_writetime(key,   col4) VALUES ('record4','DDDD') USING TTL 4000 AND TIMESTAMP 1087383600000000; -- 11:00:00
+
+-- record5: this is record 3, but to test changed ttl
+INSERT INTO source.smoke_ttl_writetime(key, t_col1) VALUES ('record5','AAAA') USING TTL   5000 AND TIMESTAMP 1087383600000000; -- 11:00:00
+INSERT INTO source.smoke_ttl_writetime(key,tw_col2) VALUES ('record5','BBBB') USING TTL  60000 AND TIMESTAMP 1087383660000000; -- 11:00:01
+INSERT INTO source.smoke_ttl_writetime(key, w_col3) VALUES ('record5','CCCC') USING TTL    400 AND TIMESTAMP 1087383720000000; -- 11:00:02
+INSERT INTO source.smoke_ttl_writetime(key,   col4) VALUES ('record5','DDDD') USING TTL 700000 AND TIMESTAMP 1087383780000000; -- 11:00:03
+
+-- record6: this is record 3, but to test changed writetime
+INSERT INTO source.smoke_ttl_writetime(key, t_col1) VALUES ('record6','AAAA') USING TTL   5000 AND TIMESTAMP 1087383600000000; -- 11:00:00
+INSERT INTO source.smoke_ttl_writetime(key,tw_col2) VALUES ('record6','BBBB') USING TTL  60000 AND TIMESTAMP 1087383660000000; -- 11:00:01
+INSERT INTO source.smoke_ttl_writetime(key, w_col3) VALUES ('record6','CCCC') USING TTL    400 AND TIMESTAMP 1087383720000000; -- 11:00:02
+INSERT INTO source.smoke_ttl_writetime(key,   col4) VALUES ('record6','DDDD') USING TTL 700000 AND TIMESTAMP 1087383780000000; -- 11:00:03
+
+-- record7: this is record 3, but to test changing both ttl and writetime
+INSERT INTO source.smoke_ttl_writetime(key, t_col1) VALUES ('record7','AAAA') USING TTL   5000 AND TIMESTAMP 1087383600000000; -- 11:00:00
+INSERT INTO source.smoke_ttl_writetime(key,tw_col2) VALUES ('record7','BBBB') USING TTL  60000 AND TIMESTAMP 1087383660000000; -- 11:00:01
+INSERT INTO source.smoke_ttl_writetime(key, w_col3) VALUES ('record7','CCCC') USING TTL    400 AND TIMESTAMP 1087383720000000; -- 11:00:02
+INSERT INTO source.smoke_ttl_writetime(key,   col4) VALUES ('record7','DDDD') USING TTL 700000 AND TIMESTAMP 1087383780000000; -- 11:00:03
+
+-- record8: this is similar to record5, but we will change t_col1 but no writetime columns, which means the change will be lost
+INSERT INTO source.smoke_ttl_writetime(key, t_col1) VALUES ('record8','AAAA') USING TTL   5000 AND TIMESTAMP 1087383600000000; -- 11:00:00
+INSERT INTO source.smoke_ttl_writetime(key,tw_col2) VALUES ('record8','BBBB') USING TTL  60000 AND TIMESTAMP 1087383660000000; -- 11:00:01
+INSERT INTO source.smoke_ttl_writetime(key, w_col3) VALUES ('record8','CCCC') USING TTL    400 AND TIMESTAMP 1087383720000000; -- 11:00:02
+INSERT INTO source.smoke_ttl_writetime(key,   col4) VALUES ('record8','DDDD') USING TTL 700000 AND TIMESTAMP 1087383780000000; -- 11:00:03
+
+
+SELECT key,
+       t_col1,
+       writetime(t_col1)  as  wt_t_col1, 
+             ttl(t_col1)  as ttl_t_col1,
+       tw_col2,
+       writetime(tw_col2) as  wt_tw_col2,
+             ttl(tw_col2) as ttl_tw_col2,
+       w_col3,
+       writetime(w_col3)  as  wt_w_col3,
+             ttl(w_col3)  as ttl_w_col3,
+       col4, 
+       writetime(col4)    as  wt_col4,
+             ttl(col4)    as ttl_col4
+  FROM source.smoke_ttl_writetime;
+
+DROP TABLE IF EXISTS target.smoke_ttl_writetime;
+-- this gc_grace_seconds is nuts but it's a single-node instance and we're trying to tombstone a deleted record immediately
+CREATE TABLE target.smoke_ttl_writetime(key text, t_col1 text,tw_col2 text,w_col3 text,col4 text, PRIMARY KEY (key)) WITH gc_grace_seconds=1;
diff --git a/SIT/smoke/03_ttl_writetime/t.out b/SIT/smoke/03_ttl_writetime/t.out
@@ -0,0 +1,8 @@
+
+key|t_col1|tw_col2|w_col3|col4|wt_col4|ttl_col
+---------+--------+---------+--------+------+-
+record3|AAAA|BBBB|CCCC|DDDD|1087383720000000|4
+record2|AAAA|BBBB|CCCC|DDDD|1087383720000000|4
+record1|AAAA|BBBB|CCCC|DDDD|1087383720000000|4
+
+(3rows)
diff --git a/SIT/test.sh b/SIT/test.sh
@@ -54,6 +54,7 @@ echo " Removing generated files:"
 echo "   ${GENERATED_FILES}"
 errors=0
 for testDir in $(ls -d ${PHASE}/*); do
+  export testDir
   for f in ${EXPECTED_FILES}; do
     if [[ ! -f ${testDir}/$f ]]; then
       _error "${testDir} is missing ${f}"
@@ -96,6 +97,7 @@ echo " Setting up tables and data (Cassandra container)"
 echo "=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-"
 errors=0
 for testDir in $(ls -d ${PHASE}/*); do
+  export testDir
   _info ${testDir} Setup tables and data 
   docker exec ${DOCKER_CASS} cqlsh -u $CASS_USERNAME -p $CASS_PASSWORD -f $testDir/setup.cql > $testDir/output/setup.out 2>$testDir/output/setup.err
   if [ $? -ne 0 ]; then
@@ -114,6 +116,7 @@ echo " Executing test (CDM container)"
 echo "=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-"
 errors=0
 for testDir in $(ls -d ${PHASE}/*); do
+  export testDir
   _info ${testDir} Executing test
   docker exec ${DOCKER_CDM} bash -e $testDir/execute.sh /$testDir > $testDir/output/execute.out 2>$testDir/output/execute.err
   if [ $? -ne 0 ]; then
@@ -132,8 +135,16 @@ echo " Checking for expected results (Cassandra container)"
 echo "=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-"
 errors=0
 for testDir in $(ls -d ${PHASE}/*); do
-  _info ${testDir} Check Expected Results 
-  docker exec ${DOCKER_CASS} cqlsh -u $CASS_USERNAME -p $CASS_PASSWORD -f $testDir/expected.cql > $testDir/output/actual.out 2>$testDir/output/actual.err
+  export testDir
+  if [ -x ${testDir}/alternateCheckResults.sh ]; then
+    _info ${testDir} Running Alternate Check Expected Results 
+    # This provides a mechanism that allows a more nuanced means of generating actual.out
+    # as not every assertion can be based on a simple "dump from cqlsh"
+    ${testDir}/alternateCheckResults.sh > $testDir/output/actual.out 2>$testDir/output/actual.err
+  else 
+    _info ${testDir} Check Expected Results 
+    docker exec ${DOCKER_CASS} cqlsh -u $CASS_USERNAME -p $CASS_PASSWORD -f $testDir/expected.cql > $testDir/output/actual.out 2>$testDir/output/actual.err
+  fi
   if [ $? -ne 0 ]; then
     _error "${testDir}/expected.cql failed, see $testDir/output/actual.out $testDir/output/and actual.err"
     errors=1
@@ -143,7 +154,7 @@ for testDir in $(ls -d ${PHASE}/*); do
   rtn=$?
   if [ $rtn -eq 1 ]; then
     _error "${testDir} files differ (expected vs actual):"
-    sdiff ${testDir}/expected.out ${testDir}/output/actual.out
+    sdiff -w 200 ${testDir}/expected.out ${testDir}/output/actual.out
     errors=1
     continue
   elif [ $rtn -ne 0 ]; then
@@ -159,6 +170,7 @@ if [ $errors -ne 0 ]; then
   _fatal "One or more expected results failed. See above ERROR(s) for details."
 fi
 
+
 _captureOutput
 
 echo
diff --git a/src/main/java/datastax/astra/migrate/DiffJobSession.java b/src/main/java/datastax/astra/migrate/DiffJobSession.java
@@ -183,6 +183,7 @@ private void diff(Record record) {
         if (!diffData.isEmpty()) {
             mismatchCounter.incrementAndGet();
             logger.error("Mismatch row found for key: {} Mismatch: {}", record.getPk(), diffData);
+            logger.info("record.pk ttl = {}, ts = {}", record.getPk().getTTL(), record.getPk().getWriteTimestamp());
 
             if (autoCorrectMismatch) {
                 writeLimiter.acquire(1);
diff --git a/src/main/java/datastax/astra/migrate/cql/EnhancedPK.java b/src/main/java/datastax/astra/migrate/cql/EnhancedPK.java
@@ -15,7 +15,7 @@ public class EnhancedPK {
     private final List<Object> values;
     private final List<MigrateDataType> types;
     private final Long writeTimestamp;
-    private final Long ttl;
+    private final Integer ttl;
     private boolean errorState = false;
     private boolean warningState = false;
     private List<String> messages;
@@ -24,7 +24,7 @@ public class EnhancedPK {
     private final Object explodeMapKey;
     private final Object explodeMapValue;
 
-    public EnhancedPK(PKFactory factory, List<Object> values, List<MigrateDataType> types, Long writeTimestamp, Long ttl, Object explodeMapKey, Object explodeMapValue) {
+    public EnhancedPK(PKFactory factory, List<Object> values, List<MigrateDataType> types, Integer ttl, Long writeTimestamp, Object explodeMapKey, Object explodeMapValue) {
         this.factory = factory;
         this.values = (null==explodeMapValue? values : new ArrayList<>(values)); // copy the list when we will modify it
         this.types = types;
@@ -39,21 +39,21 @@ public EnhancedPK(PKFactory factory, List<Object> values, List<MigrateDataType>
         validate();
     }
 
-    public EnhancedPK(PKFactory factory, List<Object> values, List<MigrateDataType> types, Long writeTimestamp, Long ttl) {
-        this(factory, values, types, writeTimestamp, ttl, null, null);
+    public EnhancedPK(PKFactory factory, List<Object> values, List<MigrateDataType> types, Integer ttl, Long writeTimestamp) {
+        this(factory, values, types, ttl, writeTimestamp, null, null);
     }
 
-    public EnhancedPK(PKFactory factory, List<Object> values, List<MigrateDataType> types, Long writeTimestamp, Long ttl, Map explodeMap) {
-        this(factory, values, types, writeTimestamp, ttl, null, null);
+    public EnhancedPK(PKFactory factory, List<Object> values, List<MigrateDataType> types, Integer ttl, Long writeTimestamp, Map<Object,Object> explodeMap) {
+        this(factory, values, types, ttl, writeTimestamp, null, null);
         this.explodeMap = explodeMap;
     }
 
     public List<EnhancedPK> explode() {
         if (null == explodeMap || explodeMap.isEmpty()) {
-            return Arrays.asList(this);
+            return Collections.singletonList(this);
         }
         return explodeMap.entrySet().stream()
-                .map(entry -> new EnhancedPK(factory, values, types, writeTimestamp, ttl, entry.getKey(), entry.getValue()))
+                .map(entry -> new EnhancedPK(factory, values, types, ttl, writeTimestamp, entry.getKey(), entry.getValue()))
                 .collect(Collectors.toList());
     }
 
@@ -71,7 +71,7 @@ public Object getExplodeMapValue() {
     public Long getWriteTimestamp() {
         return this.writeTimestamp;
     }
-    public Long getTTL() {
+    public Integer getTTL() {
         return this.ttl;
     }
 
diff --git a/src/main/java/datastax/astra/migrate/cql/PKFactory.java b/src/main/java/datastax/astra/migrate/cql/PKFactory.java
diff --git a/src/main/java/datastax/astra/migrate/cql/statements/AbstractOriginSelectStatement.java b/src/main/java/datastax/astra/migrate/cql/statements/AbstractOriginSelectStatement.java
diff --git a/src/main/java/datastax/astra/migrate/cql/statements/AbstractTargetUpsertStatement.java b/src/main/java/datastax/astra/migrate/cql/statements/AbstractTargetUpsertStatement.java
diff --git a/src/main/java/datastax/astra/migrate/cql/statements/TargetInsertStatement.java b/src/main/java/datastax/astra/migrate/cql/statements/TargetInsertStatement.java
diff --git a/src/main/java/datastax/astra/migrate/cql/statements/TargetUpdateStatement.java b/src/main/java/datastax/astra/migrate/cql/statements/TargetUpdateStatement.java
diff --git a/src/test/java/datastax/astra/migrate/cql/features/TTLAndWritetimeTest.java b/src/test/java/datastax/astra/migrate/cql/features/TTLAndWritetimeTest.java

Original file line number	Diff line number	Diff line change
`@@ -0,0 +1,3 @@`
	`1`	`+migrateData datastax.astra.migrate.Migrate migrate.properties`
	`2`	`+validateData datastax.astra.migrate.DiffData migrate.properties`
	`3`	`+fixData datastax.astra.migrate.DiffData fix.properties`
Original file line number	Diff line number	Diff line change
`@@ -0,0 +1 @@`
	`1`	`+SELECT key,t_col1,tw_col2,w_col3,col4, writetime(col4) as wt_col4, ttl(col4) as ttl_col4 FROM target.smoke_ttl_writetime;`