Address PR comments (remove caching facilities, improve conversion fu…

…nctions)
geotrellis · jpolchlo · Mar 22, 2019 · Mar 1, 2019 · Mar 1, 2019 · Mar 1, 2019
commit c4f14f43a29fb5fa658c33cc2edb01cdb238f5c6
diff --git a/src/main/scala/vectorpipe/ProcessOSM.scala b/src/main/scala/vectorpipe/ProcessOSM.scala
@@ -20,7 +20,7 @@ import com.vividsolutions.jts.{geom => jts}
 import vectorpipe.functions.osm._
 import vectorpipe.relations.MultiPolygons
 import vectorpipe.relations.Routes
-import vectorpipe.util.{Caching, Resource}
+import vectorpipe.util.Resource
 import spray.json._
 
 object ProcessOSM {
@@ -66,20 +66,6 @@ object ProcessOSM {
 
   lazy val VersionedElementEncoder: Encoder[Row] = RowEncoder(VersionedElementSchema)
 
-  /* support UDF for constructGeometries */
-  private case class StrMember(`type`: String, ref: Long, role: String)
-  private val convertMembers = org.apache.spark.sql.functions.udf { member: Seq[Row] =>
-    if (member == null)
-      null
-    else {
-      member.map { row: Row =>
-        StrMember(vectorpipe.model.Member.stringFromByte(row.getAs[Byte]("type")),
-                  row.getAs[Long]("ref"),
-                  row.getAs[String]("role"))
-      }
-    }
-  }
-
   /**
     * Snapshot pre-processed elements.
     *
@@ -232,25 +218,7 @@ object ProcessOSM {
     import input.sparkSession.implicits._
     val st_pointToGeom = org.apache.spark.sql.functions.udf { pt: jts.Point => pt.asInstanceOf[jts.Geometry] }
 
-    // The following type conversion needed in case input comes from Change source
-    val elements =
-      Try(input.schema
-               .find(_.name == "members")
-               .get
-               .dataType
-               .asInstanceOf[ArrayType]
-               .elementType
-               .asInstanceOf[StructType]
-               .find(_.name == "type")
-               .get) match {
-        case Failure(_) => throw new IllegalArgumentException(s"Could not get type of 'members' in schema ${input.schema}")
-        case Success(field) =>
-          if (field.dataType == ByteType)
-            input.withColumn("members", convertMembers('members))
-          else
-            input
-      }
-
+    val elements = elaborateMemberTypes(input)
     val nodes = ProcessOSM.preprocessNodes(elements)
 
     val nodeGeoms = ProcessOSM.constructPointGeometries(nodes)
@@ -308,8 +276,7 @@ object ProcessOSM {
     * @param _nodesToWays Optional lookup table.
     * @return Way geometries.
     */
-  def reconstructWayGeometries(_ways: DataFrame, _nodes: DataFrame, _nodesToWays: Option[DataFrame] = None)(implicit
-                                                                                                            cache: Caching = Caching.none, cachePartitions: Option[Int] = None): DataFrame = {
+  def reconstructWayGeometries(_ways: DataFrame, _nodes: DataFrame, _nodesToWays: Option[DataFrame] = None): DataFrame = {
     implicit val ss: SparkSession = _ways.sparkSession
     import ss.implicits._
     ss.withJTS
@@ -504,20 +471,14 @@ object ProcessOSM {
     * @param geoms      DataFrame containing way geometries to use in reconstruction.
     * @return Relations geometries.
     */
-  def reconstructRelationGeometries(_relations: DataFrame, geoms: DataFrame)(implicit cache: Caching = Caching.none,
-                                                                             cachePartitions: Option[Int] = None)
-  : DataFrame = {
+  def reconstructRelationGeometries(_relations: DataFrame, geoms: DataFrame): DataFrame = {
     val relations = preprocessRelations(_relations)
 
     reconstructMultiPolygonRelationGeometries(relations, geoms)
       .union(reconstructRouteRelationGeometries(relations, geoms))
   }
 
-  def reconstructMultiPolygonRelationGeometries(_relations: DataFrame, geoms: DataFrame)(implicit cache: Caching =
-  Caching.none,
-                                                                                         cachePartitions: Option[Int]
-                                                                                         = None)
-  : DataFrame = {
+  def reconstructMultiPolygonRelationGeometries(_relations: DataFrame, geoms: DataFrame): DataFrame = {
     implicit val ss: SparkSession = _relations.sparkSession
     import ss.implicits._
     ss.withJTS
@@ -579,10 +540,7 @@ object ProcessOSM {
         'minorVersion)
   }
 
-  def reconstructRouteRelationGeometries(_relations: DataFrame, geoms: DataFrame)(implicit cache: Caching = Caching
-    .none,
-                                                                                  cachePartitions: Option[Int] = None)
-  : DataFrame = {
+  def reconstructRouteRelationGeometries(_relations: DataFrame, geoms: DataFrame): DataFrame = {
     implicit val ss: SparkSession = _relations.sparkSession
     import ss.implicits._
     ss.withJTS

diff --git a/src/main/scala/vectorpipe/functions/osm/package.scala b/src/main/scala/vectorpipe/functions/osm/package.scala
@@ -1,11 +1,14 @@
 package vectorpipe.functions
 
 import org.apache.spark.sql.expressions.UserDefinedFunction
+import org.apache.spark.sql.DataFrame
 import org.apache.spark.sql.functions._
 import org.apache.spark.sql.types._
 import org.apache.spark.sql.{Column, Row}
 import vectorpipe.ProcessOSM._
+import vectorpipe.model.Member
 
+import scala.util.{Try, Success, Failure}
 import scala.util.matching.Regex
 
 package object osm {
@@ -169,11 +172,7 @@ package object osm {
 
   private val _compressMemberTypes = (members: Seq[Row]) =>
     members.map { row =>
-      val t = row.getAs[String]("type") match {
-        case "node" => NodeType
-        case "way" => WayType
-        case "relation" => RelationType
-      }
+      val t = Member.typeFromString(row.getAs[String]("type"))
       val ref = row.getAs[Long]("ref")
       val role = row.getAs[String]("role")
 
@@ -182,6 +181,37 @@ package object osm {
 
   lazy val compressMemberTypes: UserDefinedFunction = udf(_compressMemberTypes, MemberSchema)
 
+  private case class StrMember(`type`: String, ref: Long, role: String)
+
+  private val convertMembers = org.apache.spark.sql.functions.udf { member: Seq[Row] =>
+    if (member == null)
+      null
+    else {
+      member.map { row: Row =>
+        StrMember(vectorpipe.model.Member.stringFromByte(row.getAs[Byte]("type")),
+                  row.getAs[Long]("ref"),
+                  row.getAs[String]("role"))
+      }
+    }
+  }
+
+  def elaborateMemberTypes(input: DataFrame): DataFrame = {
+    // The following type conversion needed in case input comes from Change source
+    Try(input.schema("members")
+             .dataType
+             .asInstanceOf[ArrayType]
+             .elementType
+             .asInstanceOf[StructType]
+             .apply("type")) match {
+      case Failure(_) => throw new IllegalArgumentException(s"Could not get type of 'members' in schema ${input.schema}")
+      case Success(field) =>
+        if (field.dataType == ByteType)
+            input.withColumn("members", convertMembers(col("members")))
+          else
+            input
+      }
+  }
+
   // matches letters or emoji (no numbers or punctuation)
   private val ContentMatcher: Regex = """[\p{L}\uD83C-\uDBFF\uDC00-\uDFFF]""".r
   private val TrailingPunctuationMatcher: Regex = """[:]$""".r

diff --git a/src/main/scala/vectorpipe/model/AugmentedDiff.scala b/src/main/scala/vectorpipe/model/AugmentedDiff.scala
@@ -31,12 +31,7 @@ object AugmentedDiff {
   def apply(sequence: Int,
             prev: Option[Feature[GTGeometry, ElementWithSequence]],
             curr: Feature[GTGeometry, ElementWithSequence]): AugmentedDiff = {
-    val `type` = curr.data.`type` match {
-      case "node"     => ProcessOSM.NodeType
-      case "way"      => ProcessOSM.WayType
-      case "relation" => ProcessOSM.RelationType
-    }
-
+    val `type` = Member.typeFromString(curr.data.`type`)
     val minorVersion = prev.map(_.data.version).getOrElse(Int.MinValue) == curr.data.version
 
     AugmentedDiff(

diff --git a/src/main/scala/vectorpipe/model/Member.scala b/src/main/scala/vectorpipe/model/Member.scala
@@ -11,6 +11,7 @@ object Member {
       case "node"     => NodeType
       case "way"      => WayType
       case "relation" => RelationType
+      case _ => null.asInstanceOf[Byte]
   }
 
   def stringFromByte(b: Byte): String = b match {

diff --git a/src/main/scala/vectorpipe/util/Caching.scala b/src/main/scala/vectorpipe/util/Caching.scala
diff --git a/src/test/scala/vectorpipe/MultiPolygonRelationReconstructionSpec.scala b/src/test/scala/vectorpipe/MultiPolygonRelationReconstructionSpec.scala
@@ -114,14 +114,8 @@ class MultiPolygonRelationReconstructionSpec extends PropSpec with TableDrivenPr
             case ((changeset, id, version, minorVersion, updated, validUntil), rows) =>
               val members = rows.toVector
               // TODO store Bytes as the type in fixtures
-              val types = members.map(_.getAs[String]("type") match {
-                case "node" => NodeType
-                case "way" => WayType
-                case "relation" => RelationType
-                case _ => null.asInstanceOf[Byte]
-              })
+              val types = members.map(Member.typeFromString(_.getAs[String]("type")))
               val roles = members.map(_.getAs[String]("role"))
-              //val geoms = members.map(_.getAs[jts.Geometry]("geom"))
               val geoms = members.map(_.getAs[jts.Geometry]("geometry"))
               val mp = build(id, version, updated, types, roles, geoms).orNull