#97 Add scala 2.13 support (#115)

* #97 Add scala 2.13 support

* #97 Bump scala 2.12 patch verison to 18

* #97 Implement using shims

Author: Zejnilovic

build.sbt

@@ -16,16 +16,17 @@
 ThisBuild / organization := "za.co.absa"
 
 lazy val scala211 = "2.11.12"
-lazy val scala212 = "2.12.12"
-lazy val spark2   = "2.4.7"
-lazy val spark32   = "3.2.1"
-lazy val spark33   = "3.3.1"
+lazy val scala212 = "2.12.18"
+lazy val scala213 = "2.13.11"
+lazy val spark2   = "2.4.8"
+lazy val spark32   = "3.2.4"
+lazy val spark33   = "3.3.2"
 
 import Dependencies._
 import SparkVersionAxis._
 
 ThisBuild / scalaVersion := scala211
-ThisBuild / crossScalaVersions := Seq(scala211, scala212)
+ThisBuild / crossScalaVersions := Seq(scala211, scala212, scala213)
 
 ThisBuild / versionScheme := Some("early-semver")
 
@@ -55,15 +56,23 @@ lazy val parent = (project in file("."))
 lazy val `sparkCommons` = (projectMatrix in file("spark-commons"))
   .settings(commonSettings: _*)
   .sparkRow(SparkVersionAxis(spark2), scalaVersions = Seq(scala211, scala212))
-  .sparkRow(SparkVersionAxis(spark32), scalaVersions = Seq(scala212))
-  .sparkRow(SparkVersionAxis(spark33), scalaVersions = Seq(scala212))
+  .sparkRow(SparkVersionAxis(spark32), scalaVersions = Seq(scala212, scala213))
+  .sparkRow(SparkVersionAxis(spark33), scalaVersions = Seq(scala212, scala213))
   .dependsOn(sparkCommonsTest % "test")
 
 lazy val sparkCommonsTest = (projectMatrix in file("spark-commons-test"))
   .settings(
     commonSettings ++ Seq(
-    name := "spark-commons-test",
-    libraryDependencies ++= sparkDependencies(spark2)
+      name := "spark-commons-test",
+      libraryDependencies ++= sparkDependencies(if (scalaVersion.value == scala211) spark2 else spark32),
+      Compile / unmanagedSourceDirectories += {
+        val sourceDir = (Compile / sourceDirectory).value
+        if (scalaVersion.value.startsWith("2.13")) {
+          sourceDir / "scala_2.13+"
+        } else {
+          sourceDir / "scala_2.13-"
+        }
+      }
     ): _*
   )
-  .jvmPlatform(scalaVersions = Seq(scala211, scala212))
+  .jvmPlatform(scalaVersions = Seq(scala211, scala212, scala213))

project/Dependencies.scala

@@ -34,8 +34,8 @@ object Dependencies {
   def sparkCommonsDependencies(sparkVersion: String): Seq[ModuleID] = {
     Seq(
       "za.co.absa.commons"  %% "commons"     % "1.0.0",
-      "za.co.absa"          %% "spark-hofs"  % "0.4.0",
-      "za.co.absa"          %% "spark-hats"  % "0.2.2"
+      "za.co.absa"          %% "spark-hofs"  % "0.5.0",
+      "za.co.absa"          %% "spark-hats"  % "0.3.0"
     ) ++
       sparkDependencies(sparkVersion)
   }

spark-commons-test/src/main/scala/za/co/absa/spark/commons/test/YarnSparkConfiguration.scala

@@ -21,37 +21,34 @@ import org.apache.hadoop.conf.Configuration
 import org.apache.hadoop.fs.{FileSystem, Path}
 import org.apache.spark.SparkConf
 import org.apache.spark.sql.SparkSession
-import za.co.absa.spark.commons.test.YarnSparkConfiguration._
 
-import scala.collection.JavaConverters.iterableAsScalaIterableConverter
 
 class YarnSparkConfiguration(confDir: String, distJarsDir: String) extends SparkTestConfig {
-
   override def master: String = "yarn"
 
   override def appName: String = super.appName + " - Yarn"
 
   override protected def builder: SparkSession.Builder = {
     super.builder
-      .config(new SparkConf().setAll(getHadoopConfigurationForSpark(confDir)))
+      .config(new SparkConf().setAll(YarnSparkConfiguration.getHadoopConfigurationForSpark(confDir)))
       .config("spark.yarn.jars", dependencies)
       .config("spark.deploy.mode", "client")
   }
 
   protected def dependencies: String = {
     //get a list of all dist jars
     val distJars = FileSystem
-      .get(getHadoopConfiguration(confDir))
+      .get(YarnSparkConfiguration.getHadoopConfiguration(confDir))
       .listStatus(new Path(distJarsDir))
       .map(_.getPath)
-    val localJars = getDepsFromClassPath("absa")
-    val currentJars = getCurrentProjectJars
+    val localJars = YarnSparkConfiguration.getDepsFromClassPath("absa")
+    val currentJars = YarnSparkConfiguration.getCurrentProjectJars
     (distJars ++ localJars ++currentJars).mkString(",")
   }
 
 }
 
-object YarnSparkConfiguration {
+object YarnSparkConfiguration extends JavaConvertersWrapper {
 
   /**
    * Gets a Hadoop configuration object from the specified hadoopConfDir parameter
@@ -73,7 +70,7 @@ object YarnSparkConfiguration {
    * @param hadoopConf Hadoop Configuration object to be converted into Spark configs
    */
   def hadoopConfToSparkMap(hadoopConf: Configuration): Map[String, String] = {
-    hadoopConf.asScala.map(entry => (s"spark.hadoop.${entry.getKey}", entry.getValue)).toMap
+    asScala(hadoopConf).map(entry => (s"spark.hadoop.${entry.getKey}", entry.getValue)).toMap
   }
 
   /**
@@ -88,7 +85,7 @@ object YarnSparkConfiguration {
    */
   def getDepsFromClassPath(inclPattern: String): Seq[String] = {
     val cl = this.getClass.getClassLoader
-    cl.asInstanceOf[java.net.URLClassLoader].getURLs.filter(c => c.toString.contains(inclPattern)).map(_.toString())
+    cl.asInstanceOf[java.net.URLClassLoader].getURLs.toSeq.filter(c => c.toString.contains(inclPattern)).map(_.toString())
   }
 
   /**
@@ -98,6 +95,7 @@ object YarnSparkConfiguration {
     val targetDir = new File(s"${System.getProperty("user.dir")}/target")
     targetDir
       .listFiles()
+      .toSeq
       .filter(f => f.getName.split("\\.").last.toLowerCase() == "jar" && f.getName.contains("original"))
       .map(_.getAbsolutePath)
   }

spark-commons-test/src/main/scala_2.13+/za/co/absa/spark/commons/test/JavaConvertersWrapper.scala

@@ -0,0 +1,25 @@
+/*
+ * Copyright 2021 ABSA Group Limited
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package za.co.absa.spark.commons.test
+
+import scala.jdk.CollectionConverters.IterableHasAsScala
+
+trait JavaConvertersWrapper {
+  def asScala[T](item: java.lang.Iterable[T]): Iterable[T] = {
+    item.asScala
+  }
+}

spark-commons-test/src/main/scala_2.13-/za/co/absa/spark/commons/test/JavaConvertersWrapper.scala

@@ -0,0 +1,25 @@
+/*
+ * Copyright 2021 ABSA Group Limited
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package za.co.absa.spark.commons.test
+
+import scala.collection.JavaConverters.iterableAsScalaIterableConverter
+
+trait JavaConvertersWrapper {
+  def asScala[T](item: java.lang.Iterable[T]): Iterable[T] = {
+    item.asScala
+  }
+}

spark-commons/src/main/scala/za/co/absa/spark/commons/utils/ExplodeTools.scala

@@ -229,7 +229,7 @@ object ExplodeTools {
     def processStruct(schema: StructType, path: Seq[String], parentCol: Option[Column]): Seq[Column] = {
       val currentField = path.head
       val isLeaf = path.lengthCompare(1) <= 0
-      val newFields = schema.fields.flatMap(field => {
+      val newFields = schema.fields.toSeq.flatMap(field => {
         if (field.name != currentField) {
           Seq(getFullFieldPath(parentCol, field.name).as(field.name))
         } else {
@@ -286,7 +286,7 @@ object ExplodeTools {
         }
       )
       val newFields2 = if (isColumnToFound) newFields else newFields :+ col(columnFrom).as(columnTo)
-      inputDf.select(newFields2: _*)
+      inputDf.select(newFields2.toSeq: _*)
     } else {
       putFieldIntoNestedStruct(inputDf, columnFrom, SchemaUtils.splitPath(columnTo), positionColumn)
     }
@@ -301,7 +301,7 @@ object ExplodeTools {
       val isLeaf = path.lengthCompare(1) <= 0
       var isFound = false
 
-      val newFields = schema.fields.flatMap(field => {
+      val newFields = schema.fields.toSeq.flatMap(field => {
         if (field.name == columnFrom) {
           // This removes the original column name (if any) and the transient column
           Nil

spark-commons/src/main/scala/za/co/absa/spark/commons/utils/JsonUtils.scala

@@ -58,7 +58,7 @@ object JsonUtils {
    */
   def getDataFrameFromJson(json: Seq[String])(implicit spark: SparkSession): DataFrame = {
     import spark.implicits._
-    spark.read.json(json.toDS)
+    spark.read.json(json.toDS())
   }
 
   /**
@@ -71,7 +71,7 @@ object JsonUtils {
    */
   def getDataFrameFromJson(json: Seq[String], schema: StructType)(implicit spark: SparkSession): DataFrame = {
     import spark.implicits._
-    spark.read.schema(schema).json(json.toDS)
+    spark.read.schema(schema).json(json.toDS())
   }
 
   /**

spark-commons/src/main/scala/za/co/absa/spark/commons/utils/SchemaUtils.scala

@@ -52,7 +52,7 @@ object SchemaUtils {
   def getAllArraySubPaths(path: String, name: String, dt: DataType): Seq[String] = {
     val currPath = appendPath(path, name)
     dt match {
-      case s: StructType => s.fields.flatMap(f => getAllArraySubPaths(currPath, f.name, f.dataType))
+      case s: StructType => s.fields.toSeq.flatMap(f => getAllArraySubPaths(currPath, f.name, f.dataType))
       case _@ArrayType(elType, _) => getAllArraySubPaths(path, name, elType) :+ currPath
       case _ => Seq()
     }

spark-commons/src/test/scala/za/co/absa/spark/commons/errorhandler/DataFrameErrorHandlerImplicitTest.scala

@@ -33,7 +33,7 @@ class DataFrameErrorHandlerImplicitTest extends AnyFunSuite with SparkTestBase {
 
   private val col1Name = "id"
   private val col2Name = "name"
-  implicit private val df = Seq((1, "John"), (2, "Jane"), (3, "Alice")).toDF(col1Name, col2Name)
+  implicit private val df: DataFrame = Seq((1, "John"), (2, "Jane"), (3, "Alice")).toDF(col1Name, col2Name)
 
   private type ResultDfRecordType = (Option[Integer], String)
 

spark-commons/src/test/scala/za/co/absa/spark/commons/errorhandler/types/ColumnOrValueTest.scala

@@ -31,7 +31,7 @@ class ColumnOrValueTest extends AnyFunSuite {
   }
 
   test("Creation of column based on its definition") {
-    val myColumn = current_date
+    val myColumn = current_date()
     val expected = ColumnOrValueForm(myColumn, Set(), None)
     val result = ColumnOrValue(myColumn)
     expected assertTo result