Fix it:test when no SparkContext is set

Author: siadat

connector/src/it/scala/com/datastax/spark/connector/datasource/CassandraCatalogTableReadSpec.scala

@@ -78,17 +78,17 @@ class CassandraCatalogTableReadSpec extends CassandraCatalogSpecBase {
   it should "handle count pushdowns" in {
     setupBasicTable()
     val request = spark.sql(s"""SELECT COUNT(*) from $defaultKs.$testTable""")
-    val reader = request
+    var factory = request
       .queryExecution
       .executedPlan
       .collectFirst {
-        case batchScanExec: BatchScanExec=> batchScanExec.readerFactory.createReader(EmptyInputPartition)
-        case adaptiveSparkPlanExec: AdaptiveSparkPlanExec => adaptiveSparkPlanExec.executedPlan.collectLeaves().collectFirst{
-          case batchScanExec: BatchScanExec=> batchScanExec.readerFactory.createReader(EmptyInputPartition)
-        }.get
+	case batchScanExec: BatchScanExec=> batchScanExec.readerFactory
+	case adaptiveSparkPlanExec: AdaptiveSparkPlanExec => adaptiveSparkPlanExec.executedPlan.collectLeaves().collectFirst{
+	  case batchScanExec: BatchScanExec=> batchScanExec.readerFactory
+	}.get
       }
 
-    reader.get.isInstanceOf[CassandraCountPartitionReader] should be (true)
+    factory.get.asInstanceOf[CassandraScanPartitionReaderFactory].isCountQuery should be (true)
     request.collect()(0).get(0) should be (101)
   }
 

connector/src/main/scala/com/datastax/spark/connector/datasource/CassandraScanPartitionReaderFactory.scala

@@ -40,10 +40,12 @@ case class CassandraScanPartitionReaderFactory(
   readConf: ReadConf,
   queryParts: CqlQueryParts) extends PartitionReaderFactory {
 
+  def isCountQuery: Boolean = queryParts.selectedColumnRefs.contains(RowCountRef)
+
   override def createReader(partition: InputPartition): PartitionReader[InternalRow] = {
 
     val cassandraPartition = partition.asInstanceOf[CassandraPartition[Any, _ <: Token[Any]]]
-    if (queryParts.selectedColumnRefs.contains(RowCountRef)) {
+    if (isCountQuery) {
       //Count Pushdown
       CassandraCountPartitionReader(
         connector,