Merge pull request #95 from jpolchlo/update/osmesa-improvements

Bring in recent changes from OSMesa

Author: jpolchlo

.travis.yml

@@ -1,7 +1,10 @@
 language: scala
 
+jdk:
+  - openjdk8
+
 scala:
-  - 2.11.12
+  - "2.11.12"
 
 sbt_args: -J-Xmx6g
 

src/main/scala/vectorpipe/functions/osm/package.scala

@@ -5,6 +5,7 @@ import org.apache.spark.sql.functions._
 import org.apache.spark.sql.types._
 import org.apache.spark.sql.{Column, DataFrame, Row}
 import vectorpipe.model.Member
+import vectorpipe.util._
 
 import scala.util.matching.Regex
 import scala.util.{Failure, Success, Try}
@@ -292,8 +293,12 @@ package object osm {
   def isWaterway(tags: Column): Column =
     array_intersects(splitDelimitedValues(tags.getItem("waterway")), lit(WaterwayValues.toArray)) as 'isWaterway
 
-  def mergeTags: UserDefinedFunction = udf {
-    (_: Map[String, String]) ++ (_: Map[String, String])
+  def mergeTags: UserDefinedFunction = udf { (a: Map[String, String], b: Map[String, String]) =>
+    mergeMaps(a.mapValues(Set(_)), b.mapValues(Set(_)))(_ ++ _).mapValues(_.mkString(";"))
+  }
+
+  val reduceTags: UserDefinedFunction = udf { tags: Iterable[Map[String, String]] =>
+    tags.map(x => x.mapValues(Set(_))).reduce((a, b) => mergeMaps(a, b)(_ ++ _)).mapValues(_.mkString(";"))
   }
 
   val array_intersects: UserDefinedFunction = udf { (a: Seq[_], b: Seq[_]) =>

src/main/scala/vectorpipe/internal/package.scala

@@ -132,8 +132,8 @@ package object internal {
     } else {
       @transient val idByVersion = Window.partitionBy('id).orderBy('version)
 
-      // when a node has been deleted, it doesn't include any tags; use a window function to retrieve the last tags
-      // present and use those
+      // when a node has been deleted, it doesn't include any tags or nds; use a window function to retrieve the last
+      // tags and nds present and use those
       history
         .where('type === "way")
         .repartition('id)
@@ -174,13 +174,6 @@ package object internal {
     } else {
       @transient val idByUpdated = Window.partitionBy('id).orderBy('version)
 
-      val frame =
-        if (hasCompressedMemberTypes(history)) {
-          history
-        } else {
-          history.withColumn("members", compressMemberTypes('members))
-        }
-
       // when an element has been deleted, it doesn't include any tags; use a window function to retrieve the last tags
       // present and use those
       history