Merge pull request #1223 from datastax/YannMoisan-SPARKC-458-scala-212-support

Yann moisan sparkc 458 scala 212 support

Author: RussellSpitzer

.gitignore

@@ -20,5 +20,7 @@ metastore_db
 .worksheet
 .idea
 .idea_modules
+*.ipr
+*.iws
 
 checkpoint

.travis.yml

@@ -7,6 +7,7 @@ sudo: required
 dist: trusty
 scala:
   - 2.11.12
+  - 2.12.10
 
 env:
  - CASSANDRA_VERSION=2.1.15

CHANGES.txt

@@ -1,3 +1,6 @@
+2.4.2
+  * Support for Scala 2.12 (SPARKC-458)
+
 2.4.1
  * Includes all up to 2.3.3
 

project/Settings.scala

@@ -45,7 +45,8 @@ object Settings extends Build {
   lazy val buildSettings = Seq(
     organization         := "com.datastax.spark",
     version in ThisBuild := currentVersion,
-    scalaVersion         := Versions.scalaVersion,
+    scalaVersion         := Versions.scala211,
+    crossScalaVersions   := Seq(Versions.scala211, Versions.scala212),
     crossVersion         := CrossVersion.binary,
     versionStatus        := Versions.status(scalaVersion.value, scalaBinaryVersion.value)
   )

project/SparkCassandraConnectorBuild.scala

@@ -39,7 +39,7 @@ object CassandraSparkBuild extends Build {
   lazy val cassandraServerProject = Project(
     id = "cassandra-server",
     base = file(namespace),
-    settings = defaultSettings ++ Seq(
+    settings = defaultSettings ++ sparkPackageSettings ++ Seq(
       libraryDependencies ++= Seq(Artifacts.cassandraServer % "it", Artifacts.airlift),
       Testing.cassandraServerClasspath := {
         (fullClasspath in IntegrationTest).value.map(_.data.getAbsoluteFile).mkString(File.pathSeparator)

project/Versions.scala

@@ -18,10 +18,11 @@ import scala.util.Properties
 object Versions {
 
 
-  lazy val scalaVersion = "2.11.12"
+  lazy val scala211 = "2.11.7"
+  lazy val scala212 = "2.12.8"
 
   /* For `scalaBinaryVersion.value outside an sbt task. */
-  lazy val scalaBinary = scalaVersion.dropRight(2)
+  lazy val scalaBinary = scala212.dropRight(2)
 
   val Akka            = "2.3.4"
   val Cassandra       = "3.11.3"

project/plugins.sbt

@@ -18,9 +18,9 @@ addSbtPlugin("com.typesafe.sbt" % "sbt-git" % "0.8.5")
 
 addSbtPlugin("com.scalapenos" % "sbt-prompt" % "1.0.0")
 
-addSbtPlugin("org.scalastyle" %% "scalastyle-sbt-plugin" % "0.8.0")
+addSbtPlugin("org.scoverage" % "sbt-scoverage" % "1.6.1")
 
-addSbtPlugin("org.scoverage" % "sbt-scoverage" % "1.0.4")
+addSbtPlugin("org.scalastyle" %% "scalastyle-sbt-plugin" % "0.8.0")
 
 //SbtAssembly 0.12.0 is included in sbt-spark-package
 resolvers += "Spark Packages Main repo" at "https://dl.bintray.com/spark-packages/maven" 

spark-cassandra-connector-embedded/scala-2.12/src/main/scala/com/datastax/spark/connector/embedded/SparkRepl.scala

@@ -0,0 +1,42 @@
+package com.datastax.spark.connector.embedded
+
+import java.io._
+import java.net.URLClassLoader
+
+import org.apache.spark.SparkConf
+import org.apache.spark.repl.{Main, SparkILoop}
+
+import scala.collection.mutable.ArrayBuffer
+import scala.tools.nsc.GenericRunnerSettings
+
+object SparkRepl {
+
+  def runInterpreter(input: String, conf: SparkConf): String = {
+    val in = new BufferedReader(new StringReader(input + "\n"))
+    val out = new StringWriter()
+    val cl = getClass.getClassLoader
+    var paths = new ArrayBuffer[String]
+    cl match {
+      case urlLoader: URLClassLoader =>
+        for (url <- urlLoader.getURLs) {
+          if (url.getProtocol == "file") {
+            paths += url.getFile
+          }
+        }
+      case _ =>
+    }
+
+    Main.conf.setAll(conf.getAll)
+    val interp = new SparkILoop(Some(in), new PrintWriter(out))
+    Main.interp = interp
+    val separator = System.getProperty("path.separator")
+    val settings = new GenericRunnerSettings(s => throw new RuntimeException(s"Scala options error: $s"))
+    settings.processArguments(List("-classpath", paths.mkString(separator)), true)
+    interp.process(settings) // Repl starts and goes in loop of R.E.P.L
+    Main.interp = null
+    Option(Main.sparkContext).foreach(_.stop())
+    System.clearProperty("spark.driver.port")
+    out.toString
+  }
+
+}

spark-cassandra-connector/scala-2.12/src/main/scala/com/datastax/spark/connector/util/Reflect.scala

@@ -0,0 +1,21 @@
+package com.datastax.spark.connector.util
+
+import scala.reflect.runtime.universe._
+
+private[connector] object Reflect {
+
+  def constructor(tpe: Type): Symbol = tpe.decl(termNames.CONSTRUCTOR)
+
+  def member(tpe: Type, name: String): Symbol = tpe.member(TermName(name))
+
+  def methodSymbol(tpe: Type): MethodSymbol = {
+    val constructors = constructor(tpe).asTerm.alternatives.map(_.asMethod)
+    val paramCount = constructors.map(_.paramLists.flatten.size).max
+    constructors.filter(_.paramLists.flatten.size == paramCount) match {
+      case List(onlyOne) => onlyOne
+      case _             => throw new IllegalArgumentException(
+        "Multiple constructors with the same number of parameters not allowed.")
+    }
+  }
+}
+

spark-cassandra-connector/src/it/java/com/datastax/spark/connector/CassandraJavaUtilTest.java

@@ -5,6 +5,10 @@
 import java.util.Map;
 import java.util.Set;
 
+import com.datastax.spark.connector.japi.CassandraJavaUtil;
+import org.hamcrest.BaseMatcher;
+import org.hamcrest.Description;
+import org.hamcrest.Matcher;
 import scala.reflect.api.TypeTags;
 
 import static com.datastax.spark.connector.japi.CassandraJavaUtil.*;
@@ -31,61 +35,71 @@
 @SuppressWarnings("unchecked")
 public class CassandraJavaUtilTest {
 
+    /**
+     * Scala refelection type tags change the string reprsentation of some types, in scala 2.11 java.lang
+     * is included, in scala 2.12 it is removed. To remove this conflict we just always remove the java.lang
+     * portion
+     */
+    private String removeJavaLang(String target) {
+        return target.replaceAll("java.lang.", "");
+    }
+
+    private final String STRING = removeJavaLang(String.class.getName());
+    private final String LIST_STRING =
+        removeJavaLang(String.format("%s[%s]", List.class.getName(), String.class.getName()));
+    private final String MAP_STRING_INT =
+        removeJavaLang(String.format("%s[%s,%s]", Map.class.getName(), String.class.getName(), Integer.class.getName()));
+    private final String LIST_SET_MAP_STRING_INT =
+        removeJavaLang(String.format("%s[%s[%s[%s,%s]]]", List.class.getName(), Set.class.getName(), Map.class.getName(), String.class.getName(), Integer.class.getName()));
+
     @Test
     public void testTypeTag1() throws Exception {
         TypeTags.TypeTag<String> tt = typeTag(String.class);
-        assertThat(tt.tpe().toString(), is(String.class.getName()));
+        assertThat(removeJavaLang(tt.tpe().toString()), is(STRING));
     }
 
     @Test
     public void testTypeTag2() throws Exception {
         TypeTags.TypeTag<List> tt1 = typeTag(List.class, String.class);
-        assertThat(tt1.tpe().toString(), is(String.format("%s[%s]",
-                List.class.getName(), String.class.getName())));
+        assertThat(removeJavaLang(removeJavaLang(tt1.tpe().toString())), is(LIST_STRING));
 
         TypeTags.TypeTag<Map> tt2 = typeTag(Map.class, String.class, Integer.class);
-        assertThat(tt2.tpe().toString(), is(String.format("%s[%s,%s]",
-                Map.class.getName(), String.class.getName(), Integer.class.getName())));
+        assertThat(removeJavaLang(removeJavaLang(tt2.tpe().toString())), is(MAP_STRING_INT));
     }
 
     @Test
     public void testTypeTag3() throws Exception {
         TypeTags.TypeTag<List> tt = typeTag(List.class, typeTag(Set.class, typeTag(Map.class, typeTag(String.class), typeTag(Integer.class))));
-        assertThat(tt.tpe().toString(), is(String.format("%s[%s[%s[%s,%s]]]",
-                List.class.getName(), Set.class.getName(), Map.class.getName(), String.class.getName(), Integer.class.getName())));
+        assertThat(removeJavaLang(tt.tpe().toString()), is(LIST_SET_MAP_STRING_INT));
     }
 
     @Test
     public void testTypeConverter1() throws Exception {
         TypeConverter<List<String>> tc = typeConverter(String.class);
-        assertThat(tc.targetTypeName(), is(String.class.getSimpleName()));
+        assertThat(removeJavaLang(tc.targetTypeName()), is(STRING));
     }
 
     @Test
     public void testTypeConverter2() throws Exception {
         TypeConverter<List<String>> tc1 = typeConverter(List.class, String.class);
-        assertThat(tc1.targetTypeName(), is(String.format("%s[%s]",
-                List.class.getName(), String.class.getSimpleName())));
+        assertThat(removeJavaLang(tc1.targetTypeName()), is(LIST_STRING));
 
         TypeConverter<Map<String, Integer>> tc2 = typeConverter(Map.class, String.class, Integer.class);
-        assertThat(tc2.targetTypeName(), is(String.format("%s[%s,%s]",
-                Map.class.getName(), String.class.getSimpleName(), Integer.class.getName())));
+        assertThat(removeJavaLang(tc2.targetTypeName()), is(MAP_STRING_INT));
 
     }
 
     @Test
     public void testTypeConverter3() throws Exception {
         TypeConverter<List> tc = typeConverter(List.class, typeTag(Set.class, typeTag(Map.class, typeTag(String.class), typeTag(Integer.class))));
-        assertThat(tc.targetTypeName(), is(String.format("%s[%s[%s[%s,%s]]]",
-                List.class.getName(), Set.class.getName(), Map.class.getName(), String.class.getSimpleName(), Integer.class.getName())));
+        assertThat(removeJavaLang(tc.targetTypeName()), is(LIST_SET_MAP_STRING_INT));
     }
 
     @Test
     public void testTypeConverter4() throws Exception {
         TypeTags.TypeTag<List> tt = typeTag(List.class, typeTag(Set.class, typeTag(Map.class, typeTag(String.class), typeTag(Integer.class))));
         TypeConverter<List> tc = typeConverter(tt);
-        assertThat(tc.targetTypeName(), is(String.format("%s[%s[%s[%s,%s]]]",
-                List.class.getName(), Set.class.getName(), Map.class.getName(), String.class.getSimpleName(), Integer.class.getName())));
+        assertThat(removeJavaLang(tc.targetTypeName()), is(LIST_SET_MAP_STRING_INT));
     }
 
     @Test