Fix index issue & add readme details

Author: pravinbhat

README.md

@@ -62,6 +62,23 @@ spark.destination.autocorrect.missing                   true|false
 spark.destination.autocorrect.mismatch                  true|false
 ```
 
+# Migrating specific partition ranges
+- You can also use the tool to migrate specific partition ranges, use class option `--class datastax.astra.migrate.MigratePartitionsFromFile` as shown below
+```
+./spark-submit --properties-file sparkConf.properties /
+--master "local[*]" /
+--class datastax.astra.migrate.MigratePartitionsFromFile cassandra-data-migrator-1.x.jar &> logfile_name.txt
+```
+
+When running in above mode the tool assumes a `partitions.csv` file to be present in the current folder in the below format, where each line (`min,max`) represents a partition-range 
+```
+-507900353496146534,-107285462027022883
+-506781526266485690,1506166634797362039
+2637884402540451982,4638499294009575633
+798869613692279889,8699484505161403540
+```
+This mode is specifically useful to processes a subset of partition-ranges that may have generated errors as a result of a previous long-running job to migrate a large table.
+
 # Additional features
 - [Counter tables](https://docs.datastax.com/en/dse/6.8/cql/cql/cql_using/useCountersConcept.html)
 - Preserve [writetimes](https://docs.datastax.com/en/dse/6.8/cql/cql/cql_reference/cql_commands/cqlSelect.html#cqlSelect__retrieving-the-datetime-a-write-occurred-p) and [TTL](https://docs.datastax.com/en/dse/6.8/cql/cql/cql_reference/cql_commands/cqlSelect.html#cqlSelect__ref-select-ttl-p)

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

@@ -13,6 +13,7 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
+import java.util.stream.IntStream;
 
 public class AbstractJobSession extends BaseJobSession {
 
@@ -156,19 +157,13 @@ public List<MigrateDataType> getTypes(String types) {
     }
 
     public int getLargestTTL(Row sourceRow) {
-        int ttl = 0;
-        for (Integer ttlCol : ttlCols) {
-            ttl = Math.max(ttl, sourceRow.getInt(selectColTypes.size() + ttlCol - 1));
-        }
-        return ttl;
+        return IntStream.range(0, ttlCols.size())
+                .map(i -> sourceRow.getInt(selectColTypes.size() + i)).max().getAsInt();
     }
 
     public long getLargestWriteTimeStamp(Row sourceRow) {
-        long writeTimestamp = 0;
-        for (Integer writeTimeStampCol : writeTimeStampCols) {
-            writeTimestamp = Math.max(writeTimestamp, sourceRow.getLong(selectColTypes.size() + ttlCols.size() + writeTimeStampCol - 1));
-        }
-        return writeTimestamp;
+        return IntStream.range(0, writeTimeStampCols.size())
+                .mapToLong(i -> sourceRow.getLong(selectColTypes.size() + ttlCols.size() + i)).max().getAsLong();
     }
 
     public BoundStatement selectFromAstra(PreparedStatement selectStatement, Row sourceRow) {

src/main/scala/datastax/astra/migrate/BaseJob.scala

@@ -4,7 +4,6 @@ import org.apache.spark.sql.SparkSession
 import org.slf4j.LoggerFactory
 
 import java.math.BigInteger
-import java.lang.Long
 
 class BaseJob extends App {
 

src/main/scala/datastax/astra/migrate/MigratePartitionsFromFile.scala

@@ -3,9 +3,7 @@ package datastax.astra.migrate
 import com.datastax.spark.connector.cql.CassandraConnector
 import org.slf4j.LoggerFactory
 
-import java.math.BigInteger
 import scala.collection.JavaConversions._
-import java.lang.Long
 
 object MigratePartitionsFromFile extends AbstractJob {