Fix compilation warnings in Scala 2.13 (#59)

Author: angelcervera

build.sbt

@@ -126,6 +126,7 @@ def generateSparkModule(sparkVersion: String): Project = {
       Compile / resourceDirectory := baseDirectory.value / pathFromModule("src/main/resources"),
       Test / scalaSource          := baseDirectory.value / pathFromModule("src/test/scala"),
       Test / resourceDirectory    := baseDirectory.value / pathFromModule("src/test/resources"),
+      Test / parallelExecution    := false,
       crossScalaVersions := sparkScalaVersions,
       enablingPublishingSettings,
       coverageConfig,

core/src/main/scala/com/acervera/osm4scala/DenseNodesIterator.scala

@@ -56,10 +56,10 @@ class DenseNodesIterator(osmosisStringTable: StringTable,
     throw new Exception("Only visible nodes are implemented.")
   }
 
-  private val idIterator = osmosisDenseNode.id.toIterator
-  private val lonIterator = osmosisDenseNode.lon.toIterator
-  private val latIterator = osmosisDenseNode.lat.toIterator
-  private val tagsIterator = osmosisDenseNode.keysVals.toIterator
+  private val idIterator = osmosisDenseNode.id.iterator
+  private val lonIterator = osmosisDenseNode.lon.iterator
+  private val latIterator = osmosisDenseNode.lat.iterator
+  private val tagsIterator = osmosisDenseNode.keysVals.iterator
 
   private var lastNode: NodeEntity = NodeEntity(0, 0, 0, Map())
 

core/src/main/scala/com/acervera/osm4scala/FromPbfFileEntitiesIterator.scala

@@ -48,7 +48,7 @@ class FromPbfFileEntitiesIterator(pbfInputStream: InputStreamSentinel) extends E
     osmEntitiesIterator.isDefined && (osmEntitiesIterator.get.hasNext || blobIterator.hasNext)
 
   override def next(): OSMEntity = {
-    val nextEntity = osmEntitiesIterator.get.next
+    val nextEntity = osmEntitiesIterator.get.next()
 
     if (!osmEntitiesIterator.get.hasNext) {
       osmEntitiesIterator = readNextBlock()
@@ -62,7 +62,7 @@ class FromPbfFileEntitiesIterator(pbfInputStream: InputStreamSentinel) extends E
     */
   private def readNextBlock() =
     if (blobIterator.hasNext) {
-      Some(EntityIterator.fromBlob(blobIterator.next._2))
+      Some(EntityIterator.fromBlob(blobIterator.next()._2))
     } else {
       None
     }

examples/counter-akka/src/main/scala/com/acervera/osm4scala/examples/counterakka/ControllerActor.scala

@@ -50,9 +50,7 @@ class ControllerActor(pbfFile: File, filterByOsmType: Option[OSMTypes.Value]) ex
 
   val startTime = System.currentTimeMillis()
 
-  override def postStop() {
-    pbfIS.close()
-  }
+  override def postStop(): Unit = pbfIS.close()
 
   override def receive = {
 
@@ -74,7 +72,7 @@ class ControllerActor(pbfFile: File, filterByOsmType: Option[OSMTypes.Value]) ex
     // Handling the count when the counter finish and process other blob if it is available.
     case CounterResponse(count) => {
       counter += count
-      nextBlob(sender)
+      nextBlob(sender())
     }
 
   }

examples/counter-akka/src/main/scala/com/acervera/osm4scala/examples/counterakka/CounterActor.scala

@@ -54,7 +54,7 @@ class CounterActor(filterType: Option[OSMTypes.Value]) extends Actor {
 
   override def receive = {
     case BlobTupleMsg(header, blob) =>
-      sender ! CounterResponse(count(blob))
+      sender() ! CounterResponse(count(blob))
   }
 
 }

spark/src/test/scala/com/acervera/osm4scala/spark/OsmPbfFormatSpec.scala

@@ -29,13 +29,12 @@ import java.io.File
 
 import org.apache.spark.sql.catalyst.InternalRow
 import org.apache.spark.sql.{DataFrame, Row, SaveMode, SparkSession, functions => fn}
-import org.scalatest.BeforeAndAfterAll
 import org.scalatest.matchers.should.Matchers
 import org.scalatest.wordspec.AnyWordSpec
 
 import scala.util.Random
 
-class OsmPbfFormatSpec extends AnyWordSpec with Matchers with BeforeAndAfterAll {
+class OsmPbfFormatSpec extends AnyWordSpec with Matchers with SparkSessionBeforeAfterAll {
 
   def withTemporalFolder(testCode: File => Any): Unit =
     testCode(
@@ -44,42 +43,28 @@ class OsmPbfFormatSpec extends AnyWordSpec with Matchers with BeforeAndAfterAll
       )
     )
 
-  val cores = 4
   val madridPath = "core/src/test/resources/com/acervera/osm4scala/Madrid.bbbike.osm.pbf"
   val monacoPath = "core/src/test/resources/com/acervera/osm4scala/monaco-latest.osm.pbf"
 
-  val sparkSession = SparkSession
-    .builder()
-    .master(s"local[$cores]")
-    .getOrCreate()
-
-  import sparkSession.implicits._
-
-  val sqlContext = sparkSession.sqlContext
-
-  def loadOsmPbf(path: String, tableName: Option[String] = None): DataFrame = {
-    val df = sqlContext.read
+  def loadOsmPbf(spark: SparkSession, path: String, tableName: Option[String] = None): DataFrame = {
+    val df = spark.sqlContext.read
       .format("osm.pbf")
       .load(path)
       .repartition(cores * 2)
-    tableName.foreach(df.createTempView)
+    tableName.foreach(df.createOrReplaceTempView)
     df
   }
 
-  override protected def afterAll(): Unit = {
-    sparkSession.close()
-  }
-
   "OsmPbfFormat" should {
 
     "parsing all only one time" in {
-      val entitiesCount = loadOsmPbf(madridPath).count()
+      val entitiesCount = loadOsmPbf(spark, madridPath).count()
       entitiesCount shouldBe 2677227
     }
 
     "parser correctly" when {
       "is parsing nodes" in {
-        val node171946 = loadOsmPbf(madridPath).filter("id == 171946").collect()(0)
+        val node171946 = loadOsmPbf(spark, madridPath).filter("id == 171946").collect()(0)
         node171946.getAs[Long]("id") shouldBe 171946L
         node171946.getAs[Byte]("type") shouldBe 0
         node171946.getAs[Double]("latitude") shouldBe (40.42125 +- 0.001)
@@ -92,7 +77,7 @@ class OsmPbfFormatSpec extends AnyWordSpec with Matchers with BeforeAndAfterAll
       }
 
       "is parsing ways" in {
-        val way3996192 = loadOsmPbf(madridPath).filter("id == 3996192").collect()(0)
+        val way3996192 = loadOsmPbf(spark, madridPath).filter("id == 3996192").collect()(0)
         way3996192.getAs[Long]("id") shouldBe 3996192L
         way3996192.getAs[Byte]("type") shouldBe 1
         way3996192.getAs[AnyRef]("latitude") should be(null)
@@ -109,7 +94,7 @@ class OsmPbfFormatSpec extends AnyWordSpec with Matchers with BeforeAndAfterAll
       }
 
       "is parsing relations" in {
-        val relation55799 = loadOsmPbf(madridPath).filter("id == 55799").collect()(0)
+        val relation55799 = loadOsmPbf(spark, madridPath).filter("id == 55799").collect()(0)
         relation55799.getAs[Long]("id") shouldBe 55799
         relation55799.getAs[Byte]("type") shouldBe 2
         relation55799.getAs[AnyRef]("latitude") should be(null)
@@ -127,7 +112,7 @@ class OsmPbfFormatSpec extends AnyWordSpec with Matchers with BeforeAndAfterAll
     }
 
     "export to other formats" in withTemporalFolder { tmpFolder =>
-      val threeExamples = loadOsmPbf(madridPath)
+      val threeExamples = loadOsmPbf(spark, madridPath)
         .filter("id == 55799 || id == 3996192 || id == 171946")
         .orderBy("id")
 
@@ -136,7 +121,7 @@ class OsmPbfFormatSpec extends AnyWordSpec with Matchers with BeforeAndAfterAll
         .format("orc")
         .save(s"${tmpFolder}/madrid/three")
 
-      val readFromOrc = sqlContext.read
+      val readFromOrc = spark.sqlContext.read
         .format("orc")
         .load(s"${tmpFolder}/madrid/three")
         .orderBy("id")
@@ -151,7 +136,10 @@ class OsmPbfFormatSpec extends AnyWordSpec with Matchers with BeforeAndAfterAll
     "execute complex queries" when {
       "using dsl" should {
         "count arrays and filter" in {
-          loadOsmPbf(madridPath)
+          val sparkStable = spark
+          import sparkStable.implicits._
+
+          loadOsmPbf(spark, madridPath)
             .withColumn("no_of_nodes", fn.size($"nodes"))
             .withColumn("no_of_relations", fn.size($"relations"))
             .withColumn("no_of_tags", fn.size($"tags"))
@@ -161,33 +149,37 @@ class OsmPbfFormatSpec extends AnyWordSpec with Matchers with BeforeAndAfterAll
         }
       }
       "using SQL" should {
-        loadOsmPbf(madridPath, Some("madrid_shows"))
-        loadOsmPbf(monacoPath, Some("monaco_shows"))
+
         "count all zebras" in {
-          sqlContext
+          loadOsmPbf(spark, madridPath, Some("madrid_shows"))
+          spark.sqlContext
             .sql("select count(*) from madrid_shows where array_contains(map_values(tags), 'zebra')")
             .show()
         }
         "extract all keys used in tags" in {
-          sqlContext
+          loadOsmPbf(spark, madridPath, Some("madrid_shows"))
+          spark.sqlContext
             .sql("select distinct explode(map_keys(tags)) as tag from madrid_shows where size(tags) > 0 order by tag")
             .show()
         }
 
         "extract unique list of types" in {
-          sqlContext
+          loadOsmPbf(spark, monacoPath, Some("monaco_shows"))
+          spark.sqlContext
             .sql("select distinct(type) as unique_types from monaco_shows order by unique_types")
             .show()
         }
 
         "extract ways with more nodes" in {
-          sqlContext
+          loadOsmPbf(spark, monacoPath, Some("monaco_shows"))
+          spark.sqlContext
             .sql("select id, size(nodes) as size_nodes from monaco_shows where type == 1 order by size_nodes desc")
             .show()
         }
 
         "extract relations" in {
-          sqlContext
+          loadOsmPbf(spark, monacoPath, Some("monaco_shows"))
+          spark.sqlContext
             .sql("select id, relations from monaco_shows where type == 2")
             .show()
         }

spark/src/test/scala/com/acervera/osm4scala/spark/SparkTestUtilities.scala

@@ -0,0 +1,82 @@
+/*
+ * The MIT License (MIT)
+ *
+ * Copyright (c) 2020 Ángel Cervera Claudio
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a copy
+ * of this software and associated documentation files (the "Software"), to deal
+ * in the Software without restriction, including without limitation the rights
+ * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+ * copies of the Software, and to permit persons to whom the Software is
+ * furnished to do so, subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be included in all
+ * copies or substantial portions of the Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+ * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+ * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+ * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+ * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+ * SOFTWARE.
+ *
+ */
+
+package com.acervera.osm4scala.spark
+
+import org.apache.spark.sql.{SQLContext, SparkSession}
+import org.scalatest.{BeforeAndAfterAll, Suite}
+
+object SparkSessionFixture {
+
+  /**
+    * Create a new spark session.
+    * Warning 1: It there is a session open, is going to use it and close it at the end of the test.
+    * Warning 2: Using it only with `in` case, don't with `should`, `when`, etc... because the finally is executed before
+    * the test. Maybe is executed concurrently.
+    *
+    * @param cores Number of cores to use.
+    * @param appName Application name to use.
+    * @param testCode Test to execute
+    */
+  def withSparkSession(cores: Int, appName: String)(testCode: (SparkSession, SQLContext) => Any): Unit = {
+    val sparkSession = SparkSession
+      .builder()
+      .appName(appName)
+      .master(s"local[$cores]")
+      .getOrCreate()
+    try {
+      testCode(sparkSession, sparkSession.sqlContext)
+    } finally {
+      sparkSession.close()
+    }
+  }
+
+}
+
+trait SparkSessionBeforeAfterAll extends BeforeAndAfterAll { this: Suite =>
+
+  val cores: Int = 4
+  val appName: String = this.getClass().getCanonicalName()
+
+  var spark: SparkSession = _
+
+  override def beforeAll(): Unit = {
+    spark =
+      SparkSession
+        .builder()
+        .appName(appName)
+        .master(s"local[$cores]")
+        .getOrCreate()
+
+    super.beforeAll()
+  }
+
+  override def afterAll() {
+    try super.afterAll()
+    finally {
+      spark.close()
+    }
+  }
+}