Merge branch 'b1.5' into b1.6

Version Change in Spark Dep Method Signature change for IntegerOverflow SPARKC-413 Conflicts: project/Versions.scala spark-cassandra-connector/src/main/scala/com/datastax/spark/connector/rdd/partitioner/CassandraPartitionGenerator.scala
christobill · Aug 12, 2016 · 4207b71 · 4207b71
2 parents eb20be8 + 14ae351
commit 4207b71
Show file tree

Hide file tree

Showing 3 changed files with 11 additions and 5 deletions.
diff --git a/...-cassandra-connector/src/it/scala/com/datastax/spark/connector/rdd/CassandraRDDSpec.scala b/...-cassandra-connector/src/it/scala/com/datastax/spark/connector/rdd/CassandraRDDSpec.scala
@@ -253,6 +253,12 @@ class CassandraRDDSpec extends SparkCassandraITFlatSpecBase {
     result should have length 3
   }
 
+  it should "not overflow on reasonable but large split_size_in_mb" in {
+    sc.cassandraTable(ks, "simple_kv")
+      .withReadConf(ReadConf(splitSizeInMB = 10000))
+      .splitSize should be (10485760000L)
+  }
+
   it should "allow to read a Cassandra table as Array of user-defined case class (nested) objects" in {
     val result = sc.cassandraTable[SampleWithNestedScalaCaseClass#InnerClass](ks, "simple_kv").collect()
     result should have length 3

diff --git a/...tor/src/main/scala/com/datastax/spark/connector/rdd/CassandraTableRowReaderProvider.scala b/...tor/src/main/scala/com/datastax/spark/connector/rdd/CassandraTableRowReaderProvider.scala
@@ -31,7 +31,7 @@ trait CassandraTableRowReaderProvider[R] {
 
   protected def splitCount: Option[Int] = readConf.splitCount
 
-  protected def splitSize: Int = readConf.splitSizeInMB * 1024 * 1024
+  protected[connector] def splitSize: Long = readConf.splitSizeInMB * 1024L * 1024L
 
   protected def fetchSize: Int = readConf.fetchSizeInRows
 

diff --git a/...main/scala/com/datastax/spark/connector/rdd/partitioner/CassandraPartitionGenerator.scala b/...main/scala/com/datastax/spark/connector/rdd/partitioner/CassandraPartitionGenerator.scala
@@ -156,10 +156,10 @@ object CassandraPartitionGenerator {
     * and therefore you don't need to specify the ones proper for the partitioner used in the
     * Cassandra cluster. */
   def apply(
-    conn: CassandraConnector,
-    tableDef: TableDef,
-    splitCount: Option[Int],
-    splitSize: Int): CassandraPartitionGenerator[V, T] = {
+      conn: CassandraConnector,
+      tableDef: TableDef,
+      splitCount: Option[Int],
+      splitSize: Long): CassandraRDDPartitioner[V, T] = {
 
     val tokenFactory = getTokenFactory(conn)
     new CassandraPartitionGenerator(conn, tableDef, splitCount, splitSize)(tokenFactory)