Add support for varint. Change code to not require update statement when there is not counter column and default primary key string nulls to blanks.

Author: Ankitp1342

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

@@ -55,11 +55,13 @@ protected CopyJobSession(CqlSession sourceSession, CqlSession astraSession, Spar
         }
         isPreserveTTLWritetime = Boolean.parseBoolean(sparkConf.get("spark.migrate.preserveTTLWriteTime", "false"));
 
-        String updateSelectMappingStr = sparkConf.get("spark.migrate.source.counterTable.update.select.index", "0");
-        for (String updateSelectIndex : updateSelectMappingStr.split(",")) {
-            updateSelectMapping.add(Integer.parseInt(updateSelectIndex));
-        }
         if (isCounterTable) {
+            String updateSelectMappingStr = sparkConf.get("spark.migrate.source.counterTable.update.select.index", "0");
+            for (String updateSelectIndex : updateSelectMappingStr.split(",")) {
+                updateSelectMapping.add(Integer.parseInt(updateSelectIndex));
+            }
+
+
             String counterTableUpdate = sparkConf.get("spark.migrate.source.counterTable.update.cql");
             astraInsertStatement = astraSession.prepare(counterTableUpdate);
         } else {
@@ -89,11 +91,15 @@ public void getDataAndInsert(BigInteger min, BigInteger max) {
                 if (batchSize == 1 || writeTimeStampFilter) {
                     for (Row sourceRow : resultSet) {
                         readLimiter.acquire(1);
-                        // only process rows greater than writeTimeStampFilter
-                        Long sourceWriteTimeStamp = getLargestWriteTimeStamp(sourceRow);
-                        if (writeTimeStampFilter && (sourceWriteTimeStamp < minWriteTimeStampFilter)
-                                || sourceWriteTimeStamp > maxWriteTimeStampFilter) {
-                            continue;
+
+                        if(writeTimeStampFilter) {
+                            // only process rows greater than writeTimeStampFilter
+                            Long sourceWriteTimeStamp = getLargestWriteTimeStamp(sourceRow);
+                            if (sourceWriteTimeStamp < minWriteTimeStampFilter
+                                    || sourceWriteTimeStamp > maxWriteTimeStampFilter) {
+                                continue;
+                            }
+
                         }
 
                         writeLimiter.acquire(1);
@@ -222,7 +228,11 @@ public BoundStatement bindInsert(PreparedStatement insertStatement, Row sourceRo
                 MigrateDataType dataType = insertColTypes.get(index);
 
                 try {
-                    boundInsertStatement = boundInsertStatement.set(index, getData(dataType, index, sourceRow), dataType.typeClass);
+                    Object colData = getData(dataType, index, sourceRow);
+                    if(index < idColTypes.size() && colData==null && dataType.typeClass==String.class){
+                        colData="";
+                    }
+                    boundInsertStatement = boundInsertStatement.set(index, colData, dataType.typeClass);
                 } catch (NullPointerException e) {
                     // ignore the exception for map values being null
                     if (dataType.typeClass != Map.class) {

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

@@ -4,6 +4,7 @@
 import com.datastax.oss.driver.api.core.data.UdtValue;
 
 import java.math.BigDecimal;
+import java.math.BigInteger;
 import java.nio.ByteBuffer;
 import java.time.Instant;
 import java.time.LocalDate;
@@ -78,6 +79,8 @@ private Class getType(int type) {
                 return LocalDate.class;
             case 16:
                 return UdtValue.class;
+            case 17:
+                return BigInteger.class;
         }
 
         return Object.class;