Merge pull request #1284 from datastax/SPARKC-612-2.5

SPARKC-612 throw a meaningful exception when reading a table view

Author: jtgrabowski

connector/src/it/scala/com/datastax/spark/connector/SparkCassandraITFlatSpecBase.scala

@@ -158,7 +158,17 @@ trait SparkCassandraITSpecBase
 
   def skipIfProtocolVersionGTE(protocolVersion: ProtocolVersion)(f: => Unit): Unit = {
     if (!(pv.getCode >= protocolVersion.getCode)) f
-    else report(s"Skipped Because ProtcolVersion $pv >= $protocolVersion")
+    else report(s"Skipped Because ProtocolVersion $pv >= $protocolVersion")
+  }
+
+  /** Skips the given test if the Cluster Version is lower or equal to the given `cassandra` Version or `dse` Version
+    * (if this is a DSE cluster) */
+  def skipIfLT(cassandra: Version, dse: Version)(f: => Unit): Unit = {
+    if (isDse(conn)) {
+      skipIfCassandraLT(dse)(f)
+    } else {
+      skipIfCassandraLT(cassandra)(f)
+    }
   }
 
   val Cass36: Version = Version.parse("3.6.0")
@@ -167,7 +177,7 @@ trait SparkCassandraITSpecBase
     val verOrd = implicitly[Ordering[Version]]
     import verOrd._
     if (cluster.getCassandraVersion >= cassandraVersion) f
-    else report(s"Skipped because Cassandra Version ${cluster.getCassandraVersion} < $cassandraVersion")
+    else report(s"Skipped because cluster Version ${cluster.getCassandraVersion} < $cassandraVersion")
   }
 
   def skipIfProtocolVersionLT(protocolVersion: ProtocolVersion)(f: => Unit): Unit = {
@@ -176,11 +186,15 @@ trait SparkCassandraITSpecBase
   }
 
   def skipIfNotDSE(connector: CassandraConnector)(f: => Unit): Unit = {
-    val firstNodeExtras = connector.withSessionDo(_.getMetadata.getNodes.values().asScala.head.getExtras)
-    if (firstNodeExtras.containsKey(DseNodeProperties.DSE_VERSION)) f
+    if (isDse(connector)) f
     else report(s"Skipped because not DSE")
   }
 
+  private def isDse(connector: CassandraConnector): Boolean = {
+    val firstNodeExtras = connector.withSessionDo(_.getMetadata.getNodes.values().asScala.head.getExtras)
+    firstNodeExtras.containsKey(DseNodeProperties.DSE_VERSION)
+  }
+
   implicit val ec = SparkCassandraITSpecBase.ec
 
   def await[T](unit: Future[T]): T = {

connector/src/it/scala/com/datastax/spark/connector/rdd/CassandraRDDSpec.scala

@@ -6,9 +6,10 @@ import java.util.Date
 
 import com.datastax.oss.driver.api.core.DefaultProtocolVersion._
 import com.datastax.oss.driver.api.core.config.DefaultDriverOption
-import com.datastax.oss.driver.api.core.cql.{BoundStatement, SimpleStatement}
+import com.datastax.oss.driver.api.core.cql.SimpleStatement
 import com.datastax.oss.driver.api.core.cql.SimpleStatement._
 import com.datastax.spark.connector._
+import com.datastax.spark.connector.ccm.CcmConfig.{V4_0_0, V6_7_0}
 import com.datastax.spark.connector.cluster.DefaultCluster
 import com.datastax.spark.connector.cql.{CassandraConnector, CassandraConnectorConf}
 import com.datastax.spark.connector.mapper.{DefaultColumnMapper, JavaBeanColumnMapper, JavaTestBean, JavaTestUDTBean}
@@ -1107,6 +1108,16 @@ class CassandraRDDSpec extends SparkCassandraITFlatSpecBase with DefaultCluster
     )
   }
 
+  it should "throw a meaningful exception when reading a table view" in skipIfLT(cassandra = V4_0_0, dse = V6_7_0) {
+    import org.apache.spark.sql.cassandra._
+
+    val ex = intercept[IllegalArgumentException] {
+      val data = spark.read.cassandraFormat("sstable_tasks", "system_views").load()
+      data.show()
+    }
+    ex.getMessage should contain("Table views are not supported")
+  }
+
   it should "throw an exception when trying to write to a Materialized View" in skipIfProtocolVersionLT(V4){
     intercept[IllegalArgumentException] {
       sc.parallelize(Seq(("US", 1, "John", "DOE", "jdoe"))).saveToCassandra(ks, "user_by_country")

connector/src/main/scala/com/datastax/spark/connector/rdd/partitioner/CassandraPartitionGenerator.scala

@@ -79,6 +79,10 @@ private[connector] class CassandraPartitionGenerator[V, T <: Token[V]](
   def partitions: Seq[CassandraPartition[V, T]] = {
     val tokenRanges = describeRing
     val endpointCount = tokenRanges.map(_.replicas).reduce(_ ++ _).size
+    if (endpointCount == 0)
+      throw new IllegalArgumentException(s"Could not retrieve endpoints for the given table " +
+        s"(${keyspaceName}.${tableDef.name}), are you trying to read a table view? Table views are not supported, " +
+        s"see SPARKC-612.")
     val maxGroupSize = tokenRanges.size / endpointCount
 
     val splitter = createTokenRangeSplitter

test-support/src/main/scala/com/datastax/spark/connector/ccm/CcmConfig.scala

@@ -116,6 +116,8 @@ object CcmConfig {
   val DEFAULT_SERVER_LOCALHOST_KEYSTORE_PATH: String = "/server_localhost.keystore"
 
   // major DSE versions
+  val V6_8_0: Version = Version.parse("6.8.0")
+  val V6_7_0: Version = Version.parse("6.7.0")
   val V6_0_0: Version = Version.parse("6.0.0")
   val V5_1_0: Version = Version.parse("5.1.0")
   val V5_0_0: Version = Version.parse("5.0.0")