Implemented PR recomendations

Author: pravinbhat

README.md

@@ -123,7 +123,7 @@ This mode is specifically useful to processes a subset of partition-ranges that
 - Supports adding custom fixed `writetime`
 
 # Known Limitations
-- This tool does not migrate `ttl` & `writetime` for every field-level (for optimization reasons), but it instead it finds the field with the highest `ttl` & `writetime` and uses those values for `ttl` & `writetime` on the entire Target row
+- This tool does not migrate `ttl` & `writetime` at the field-level (for optimization reasons). It instead finds the field with the highest `ttl` & the field with the highest `writetime` within an `origin` row and uses those values on the entire `target` row.
 
 # Building Jar for local development
 1. Clone this repo

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

@@ -32,8 +32,8 @@ protected GuardrailJobSession(CqlSession session, SparkConf sc) {
                 sourceKeyspaceTable.split("\\.")[1], Util.getSparkPropOrEmpty(sc, "spark.query.origin"));
         String selectCols = String.join(",", tableInfo.getAllColumns());
         String partitionKey = String.join(",", tableInfo.getPartitionKeyColumns());
-        String originSelectQry = "select " + selectCols + " from " + sourceKeyspaceTable +
-                " where token(" + partitionKey + ") >= ? and token(" + partitionKey + ") <= ?  " +
+        String originSelectQry = "SELECT " + selectCols + " FROM " + sourceKeyspaceTable +
+                " WHERE TOKEN(" + partitionKey + ") >= ? AND TOKEN(" + partitionKey + ") <= ?  " +
                 sourceSelectCondition + " ALLOW FILTERING";
 
         logger.info("PARAM -- Detected Table Schema: {}", tableInfo);

src/main/java/datastax/astra/migrate/schema/ColumnInfo.java

@@ -5,6 +5,7 @@
 import lombok.Data;
 
 import java.util.List;
+import java.util.Locale;
 import java.util.Map;
 import java.util.Set;
 
@@ -32,7 +33,7 @@ public class ColumnInfo {
         if (typeInfo.getTypeClass().equals(UdtValue.class)) {
             isUDT = true;
         }
-        if (cm.getType().toString().contains(", frozen")) {
+        if (cm.getType().toString().toLowerCase(Locale.ROOT).contains(", frozen")) {
             isFrozen = true;
         }
     }