Merge pull request #1286 from datastax/SPARKC-621-2.5

SPARKC-621 SAI support

Author: jtgrabowski

.travis.yml

@@ -32,7 +32,7 @@ env:
     - CCM_CASSANDRA_VERSION=3.0.20
     - CCM_CASSANDRA_VERSION=2.1.21
     - CCM_CASSANDRA_VERSION=6.0.11 CCM_IS_DSE=true DSE_VERSION=6.0.11 CCM_SERVER_TYPE=dse
-    - CCM_CASSANDRA_VERSION=6.7.7  CCM_IS_DSE=true DSE_VERSION=6.7.7 CCM_SERVER_TYPE=dse
+    - CCM_CASSANDRA_VERSION=6.8.5  CCM_IS_DSE=true DSE_VERSION=6.8.5 CCM_SERVER_TYPE=dse
 
 install:
   - pip install --user ccm

connector/src/it/scala/com/datastax/bdp/spark/search/SearchAnalyticsIntegrationSpec.scala

@@ -42,7 +42,7 @@ class SearchAnalyticsIntegrationSpec extends SparkCassandraITFlatSpecBase with D
     val typeIndex: String = s"$ks.$typesTable"
     val weirdIndex: String = s"$ks.$weirdTable"
 
-    skipIfNotDSE(conn) {
+    dseOnly {
       conn.withSessionDo { session =>
 
         val executor = getExecutor(session)
@@ -159,7 +159,7 @@ class SearchAnalyticsIntegrationSpec extends SparkCassandraITFlatSpecBase with D
   }
 
 
-  "SearchAnalytics" should "be able to use solr_query" in skipIfNotDSE(conn) {
+  "SearchAnalytics" should "be able to use solr_query" in dseOnly {
     val df = spark.sql(s"""SELECT key,a,b,c FROM $table WHERE solr_query = '{"q": "*:*", "fq":["key:[500 TO *]", "a:1"]}' """)
 
     //Check that the correct data got through
@@ -174,7 +174,7 @@ class SearchAnalyticsIntegrationSpec extends SparkCassandraITFlatSpecBase with D
     results.size should be(expected.size)
   }
 
-  it should "be able to use solr optimizations " in skipIfNotDSE(conn) {
+  it should "be able to use solr optimizations " in dseOnly {
     val df = spark.sql(s"SELECT key,a,b,c FROM $table WHERE key >= 500 AND a == 1")
 
     //Check that pushdown happened
@@ -196,7 +196,7 @@ class SearchAnalyticsIntegrationSpec extends SparkCassandraITFlatSpecBase with D
     results.size should be(expected.size)
   }
 
-  it should "not use solr optimizations if there is a manual optimization" in skipIfNotDSE(conn) {
+  it should "not use solr optimizations if there is a manual optimization" in dseOnly {
     val df = spark.sql(s"SELECT key,a,b,c FROM $table WHERE solr_query='key:[500 TO *]' AND a == 1")
     val whereClause = df.getUnderlyingCqlWhereClause()
     whereClause.predicates.head should include("solr_query")
@@ -217,7 +217,7 @@ class SearchAnalyticsIntegrationSpec extends SparkCassandraITFlatSpecBase with D
 
   }
 
-  it should "should choose solr clauses over clustering key clauses " in skipIfNotDSE(conn) {
+  it should "should choose solr clauses over clustering key clauses " in dseOnly {
     val df = spark.sql(s"SELECT key,a,b,c FROM $table WHERE a > 2")
 
     //Check that pushdown happened
@@ -236,7 +236,7 @@ class SearchAnalyticsIntegrationSpec extends SparkCassandraITFlatSpecBase with D
     results.size should be(expected.size)
   }
 
-  it should "should be able to turn off solr optimization " in skipIfNotDSE(conn) {
+  it should "should be able to turn off solr optimization " in dseOnly {
     val df = spark.sql(s"SELECT key,a,b,c FROM $tableNoSolr WHERE a > 2")
 
     //Check that solr pushdown didn't happened
@@ -247,7 +247,7 @@ class SearchAnalyticsIntegrationSpec extends SparkCassandraITFlatSpecBase with D
     results.size should be(3000)
   }
 
-  it should "should correctly do negation filters " in skipIfNotDSE(conn) {
+  it should "should correctly do negation filters " in dseOnly {
     val df = spark.sql(
       s"""SELECT key,a,b,c FROM $table WHERE
          |key > 998 AND
@@ -270,7 +270,7 @@ class SearchAnalyticsIntegrationSpec extends SparkCassandraITFlatSpecBase with D
     results should contain theSameElementsAs (expected)
   }
 
-  it should "handle OR conjunctions when possible" in skipIfNotDSE(conn) {
+  it should "handle OR conjunctions when possible" in dseOnly {
     val df = spark.sql(
       s"""SELECT key,a,b,c FROM $table
          |WHERE key < 10
@@ -295,7 +295,7 @@ class SearchAnalyticsIntegrationSpec extends SparkCassandraITFlatSpecBase with D
     results should contain theSameElementsAs (expected)
   }
 
-  it should "handle conjunctions with LIKE clauses" in skipIfNotDSE(conn) {
+  it should "handle conjunctions with LIKE clauses" in dseOnly {
     val df = spark.sql(
       s"""SELECT key,a,b,c FROM $table
          |WHERE c LIKE '%100%'
@@ -322,7 +322,7 @@ class SearchAnalyticsIntegrationSpec extends SparkCassandraITFlatSpecBase with D
     results should contain theSameElementsAs (expected)
   }
 
-  it should "handle IN clauses" in skipIfNotDSE(conn) {
+  it should "handle IN clauses" in dseOnly {
     val df = spark.sql(
       s"""SELECT key,a,b,c FROM $table
          |WHERE key IN (4, 8, 15, 16, 23, 42)""".stripMargin)
@@ -342,7 +342,7 @@ class SearchAnalyticsIntegrationSpec extends SparkCassandraITFlatSpecBase with D
     results should contain theSameElementsAs (expected)
   }
 
-  it should "handle IsNotNull on a partition key" in skipIfNotDSE(conn) {
+  it should "handle IsNotNull on a partition key" in dseOnly {
     val df = spark.sql(
       s"""SELECT key,a,b,c FROM $table WHERE key IS NOT NULL""")
 
@@ -358,7 +358,7 @@ class SearchAnalyticsIntegrationSpec extends SparkCassandraITFlatSpecBase with D
     results should contain theSameElementsAs (expected)
   }
 
-  it should "handle mixed partition key and solr restrictions" in skipIfNotDSE(conn) {
+  it should "handle mixed partition key and solr restrictions" in dseOnly {
     val df = spark.sql(
       s"""SELECT key,a,b,c FROM $table WHERE key = 4 AND b = 5""")
 
@@ -378,7 +378,7 @@ class SearchAnalyticsIntegrationSpec extends SparkCassandraITFlatSpecBase with D
     results should contain theSameElementsAs (expected)
   }
 
-  it should "handle isNotNull all by itself" in skipIfNotDSE(conn) {
+  it should "handle isNotNull all by itself" in dseOnly {
     val df = spark.sql(
       s"""SELECT key,a,b,c FROM $table WHERE c IS NOT NULL""")
 
@@ -390,7 +390,7 @@ class SearchAnalyticsIntegrationSpec extends SparkCassandraITFlatSpecBase with D
     df.collect()
   }
 
-  it should "handle only partition key restrictions" in skipIfNotDSE(conn) {
+  it should "handle only partition key restrictions" in dseOnly {
     val df = spark.sql(
       s"""SELECT key,a,b,c FROM $table WHERE key = 4""")
 
@@ -407,7 +407,7 @@ class SearchAnalyticsIntegrationSpec extends SparkCassandraITFlatSpecBase with D
     results should contain theSameElementsAs (expected)
   }
 
-  it should "handle primary key restrictions" in skipIfNotDSE(conn) {
+  it should "handle primary key restrictions" in dseOnly {
     val df = spark.sql(
       s"""SELECT key,a,b,c FROM $table WHERE key = 4 AND a > 2""")
 
@@ -425,7 +425,7 @@ class SearchAnalyticsIntegrationSpec extends SparkCassandraITFlatSpecBase with D
     results should contain theSameElementsAs (expected)
   }
 
-  it should "handle Primary key restrictions and solr queries at the same time" in skipIfNotDSE(conn) {
+  it should "handle Primary key restrictions and solr queries at the same time" in dseOnly {
     val df = spark.sql(
       s"""SELECT key,a,b,c FROM $table WHERE key = 4 AND a > 2 AND b < 25""")
     //Check that pushdown happened
@@ -443,7 +443,7 @@ class SearchAnalyticsIntegrationSpec extends SparkCassandraITFlatSpecBase with D
     results should contain theSameElementsAs (expected)
   }
 
-  it should "optimize a count(*) without any predicates in " in skipIfNotDSE(conn) {
+  it should "optimize a count(*) without any predicates in " in dseOnly {
     val df = spark.sql(s"SELECT COUNT(*) from $table")
     val whereClause = df.getUnderlyingCqlWhereClause()
     val predicates = whereClause.predicates.head
@@ -461,63 +461,63 @@ class SearchAnalyticsIntegrationSpec extends SparkCassandraITFlatSpecBase with D
     df.collect() should have size 1
   }
 
-  it should "work with ascii" in skipIfNotDSE(conn) {
+  it should "work with ascii" in dseOnly {
     testType("""a = 'one:\".'""", "a:one")
   }
 
-  it should "work with solr reserved words in ascii" in skipIfNotDSE(conn) {
+  it should "work with solr reserved words in ascii" in dseOnly {
     testType("a = 'OR'", "a:\\\\OR")
   }
 
-  it should "work with bigint" in skipIfNotDSE(conn) {
+  it should "work with bigint" in dseOnly {
     testType("b = 1", "b:1")
   }
 
-  it should "work with date" in skipIfNotDSE(conn) {
+  it should "work with date" in dseOnly {
     testType("d = cast('2017-2-7' as date)", """d:2017\\-02\\-07""")
   }
 
-  it should "work with decimal" in skipIfNotDSE(conn) {
+  it should "work with decimal" in dseOnly {
     testType("e > 1.0", "1.000000000000000000 TO *")
   }
 
-  it should "work with double" in skipIfNotDSE(conn) {
+  it should "work with double" in dseOnly {
     testType("f = 2.0", "f:2.0")
   }
 
-  it should "work with inet" in skipIfNotDSE(conn) {
+  it should "work with inet" in dseOnly {
     testType("g = '8.8.8.8'", "g:8.8.8.8")
   }
 
-  it should "work with int" in skipIfNotDSE(conn) {
+  it should "work with int" in dseOnly {
     testType("h = 1", "h:1")
   }
 
-  it should "work with text" in skipIfNotDSE(conn) {
+  it should "work with text" in dseOnly {
     testType("i = 'one'", "i:one")
   }
 
-  it should "work with solr reserved words in text" in skipIfNotDSE(conn) {
+  it should "work with solr reserved words in text" in dseOnly {
     testType("i = 'OR'", "i:\\\\OR")
   }
 
-  it should "work with timestamp" in skipIfNotDSE(conn) {
+  it should "work with timestamp" in dseOnly {
     testType("j < cast( \"2000-01-01T00:08:20.000Z\" as timestamp)", """{"q":"*:*", "fq":["j:[* TO 2000\\-01\\-01T00\\:08\\:20Z}"]}""")
   }
 
-  it should "work with varchar" in skipIfNotDSE(conn) {
+  it should "work with varchar" in dseOnly {
     testType("k = 'one'", "k:one")
   }
 
-  it should "work with varint" in skipIfNotDSE(conn) {
+  it should "work with varint" in dseOnly {
     testType("l = 1", "l:1")
   }
 
-  it should "work with uuid" in skipIfNotDSE(conn) {
+  it should "work with uuid" in dseOnly {
     testType("m = '10000000-0000-0000-0000-000000000000'", """"m:10000000\\-0000\\-0000\\-0000\\-000000000000"""")
   }
 
-  it should "work with weird column names" in skipIfNotDSE(conn) {
+  it should "work with weird column names" in dseOnly {
     val df: DataFrame = spark
       .read
       .cassandraFormat(weirdTable, ks)
@@ -538,14 +538,14 @@ class SearchAnalyticsIntegrationSpec extends SparkCassandraITFlatSpecBase with D
     rows.head.getAs[String]("MixEdCol") should be("world")
   }
 
-  "Automatic Solr Optimization" should "occur when the selected amount of data is less than the threshold" in skipIfNotDSE(conn) {
+  "Automatic Solr Optimization" should "occur when the selected amount of data is less than the threshold" in dseOnly {
     val df = spark.sql(s"SELECT COUNT(*) from $tableAutoSolr where key < 5")
     val whereClause = df.getUnderlyingCqlWhereClause()
     val predicates = whereClause.predicates.head
     predicates should include("""solr_query""")
   }
 
-  it should "not occur if the selected amount of data is greater than the threshold" in skipIfNotDSE(conn) {
+  it should "not occur if the selected amount of data is greater than the threshold" in dseOnly {
     val df = spark.sql(s"SELECT COUNT(*) from $tableAutoSolr where key < 500")
     val whereClause = df.getUnderlyingCqlWhereClause()
     val predicates = whereClause.predicates

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

@@ -28,7 +28,6 @@ import scala.concurrent.{Await, ExecutionContext, Future}
 import scala.util.Try
 
 trait SparkCassandraITFlatSpecBase extends FlatSpec with SparkCassandraITSpecBase {
-  override def report(message: String): Unit = info
 }
 
 trait SparkCassandraITWordSpecBase extends WordSpec with SparkCassandraITSpecBase {
@@ -42,7 +41,8 @@ trait SparkCassandraITSpecBase
   with Matchers
   with BeforeAndAfterAll
   with ClusterProvider
-  with Logging {
+  with Logging
+  with Alerting {
 
   final def defaultConf: SparkConf = {
     SparkTemplate.defaultConf
@@ -150,9 +150,9 @@ trait SparkCassandraITSpecBase
     }
   }
 
-  def pv = conn.withSessionDo(_.getContext.getProtocolVersion)
+  def pv: ProtocolVersion = conn.withSessionDo(_.getContext.getProtocolVersion)
 
-  def report(message: String): Unit = {}
+  def report(message: String): Unit = alert(message)
 
   val ks = getKsName
 
@@ -161,33 +161,50 @@ trait SparkCassandraITSpecBase
     else report(s"Skipped Because ProtocolVersion $pv >= $protocolVersion")
   }
 
+  def skipIfProtocolVersionLT(protocolVersion: ProtocolVersion)(f: => Unit): Unit = {
+    if (!(pv.getCode < protocolVersion.getCode)) f
+    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 = {
+  def from(cassandra: Version, dse: Version)(f: => Unit): Unit = {
     if (isDse(conn)) {
-      skipIfCassandraLT(dse)(f)
+      from(dse)(f)
     } else {
-      skipIfCassandraLT(cassandra)(f)
+      from(cassandra)(f)
     }
   }
 
-  val Cass36: Version = Version.parse("3.6.0")
+  /** Skips the given test if the Cluster Version is lower or equal to the given version */
+  def from(version: Version)(f: => Unit): Unit = {
+    skip(cluster.getCassandraVersion, version) { f }
+  }
 
-  def skipIfCassandraLT(cassandraVersion: Version)(f: => Unit): Unit = {
-    val verOrd = implicitly[Ordering[Version]]
-    import verOrd._
-    if (cluster.getCassandraVersion >= cassandraVersion) f
-    else report(s"Skipped because cluster Version ${cluster.getCassandraVersion} < $cassandraVersion")
+  /** Skips the given test if the cluster is not DSE */
+  def dseOnly(f: => Unit): Unit = {
+    if (isDse(conn)) f
+    else report(s"Skipped because not DSE")
   }
 
-  def skipIfProtocolVersionLT(protocolVersion: ProtocolVersion)(f: => Unit): Unit = {
-    if (!(pv.getCode < protocolVersion.getCode)) f
-    else report(s"Skipped Because ProtocolVersion $pv < $protocolVersion")
+  /** Skips the given test if the cluster is not Cassandra */
+  def cassandraOnly(f: => Unit): Unit = {
+    if (isDse(conn)) report(s"Skipped because not Cassandra")
+    else f
   }
 
-  def skipIfNotDSE(connector: CassandraConnector)(f: => Unit): Unit = {
-    if (isDse(connector)) f
-    else report(s"Skipped because not DSE")
+  /** Skips the given test if the Cluster Version is lower or equal to the given version or the cluster is not DSE */
+  def dseFrom(version: Version)(f: => Any): Unit = {
+    dseOnly {
+      skip(cluster.getDseVersion.get, version) { f }
+    }
+  }
+
+  private def skip(clusterVersion: Version, minVersion: Version)(f: => Unit): Unit = {
+    val verOrd = implicitly[Ordering[Version]]
+    import verOrd._
+    if (clusterVersion >= minVersion) f
+    else report(s"Skipped because cluster Version ${cluster.getCassandraVersion} < $minVersion")
   }
 
   private def isDse(connector: CassandraConnector): Boolean = {

connector/src/it/scala/com/datastax/spark/connector/cql/ContinuousPagingScannerSpec.scala

@@ -100,7 +100,7 @@ class ContinuousPagingScannerSpec extends SparkCassandraITFlatSpecBase with Defa
     withClue(sessions.map(_.toString).mkString("\n"))(sessions.size should be(1))
   }
 
-  it should "apply MB/s throughput limit" in skipIfNotDSE(conn) {
+  it should "apply MB/s throughput limit" in dseOnly {
     val readConf = ReadConf(throughputMiBPS = Some(32.0))
     val executedStmt = executeContinuousPagingScan(readConf)
 
@@ -109,7 +109,7 @@ class ContinuousPagingScannerSpec extends SparkCassandraITFlatSpecBase with Defa
     executedStmt.getExecutionProfile.getInt(DseDriverOption.CONTINUOUS_PAGING_PAGE_SIZE) should be(33554) // 32MB/s
   }
 
-  it should "apply reads/s throughput limit" in skipIfNotDSE(conn) {
+  it should "apply reads/s throughput limit" in dseOnly {
     val readConf = ReadConf(fetchSizeInRows = 999, readsPerSec = Some(5))
     val executedStmt = executeContinuousPagingScan(readConf)
 
@@ -118,7 +118,7 @@ class ContinuousPagingScannerSpec extends SparkCassandraITFlatSpecBase with Defa
     executedStmt.getExecutionProfile.getInt(DseDriverOption.CONTINUOUS_PAGING_PAGE_SIZE) should be(999)
   }
 
-  it should "throw a meaningful exception when pages per second does not fall int (0, Int.MaxValue)" in skipIfNotDSE(conn) {
+  it should "throw a meaningful exception when pages per second does not fall int (0, Int.MaxValue)" in dseOnly {
     val readConfs = Seq(
       ReadConf(throughputMiBPS = Some(1.0 + Int.MaxValue), readsPerSec = Some(1)),
       ReadConf(throughputMiBPS = Some(-1)),