{"action":"created","comment":{"url":"https://api.github.com/repos/geotrellis/vectorpipe/pulls/comments/191756652","pull_request_review_id":124389811,"id":191756652,"diff_hunk":"@@ -0,0 +1,767 @@\n+package vectorpipe.osm.internal\n+\n+import java.io._\n+import java.sql.Timestamp\n+\n+import org.locationtech.geomesa.spark.jts._\n+import org.apache.spark.sql.jts.GeometryUDT\n+\n+import com.vividsolutions.jts.geom._\n+\n+import geotrellis.vector.{Extent, GeomFactory, SpatialIndex}\n+//import geotrellis.vector.io._\n+//import geotrellis.vector.io.json._\n+import org.apache.log4j.Logger\n+import org.apache.spark.sql._\n+import org.apache.spark.sql.catalyst.encoders.RowEncoder\n+import org.apache.spark.sql.catalyst.expressions.GenericRowWithSchema\n+import org.apache.spark.sql.expressions.Window\n+import org.apache.spark.sql.functions._\n+import org.apache.spark.sql.types._\n+\n+import vectorpipe.osm.internal.functions._\n+import vectorpipe.osm.internal.util.Caching\n+\n+import spray.json._\n+\n+object ProcessOSM {\n+  val NodeType: Byte = 1\n+  val WayType: Byte = 2\n+  val RelationType: Byte = 3\n+  val MultiPolygonRoles: Seq[String] = Set(\"\", \"outer\", \"inner\").toSeq\n+\n+  lazy val logger: Logger = Logger.getLogger(getClass)\n+\n+  val BareElementSchema = StructType(\n+    StructField(\"changeset\", LongType, nullable = false) ::\n+      StructField(\"id\", LongType, nullable = false) ::\n+      StructField(\"version\", IntegerType, nullable = false) ::\n+      StructField(\"updated\", TimestampType, nullable = false) ::\n+      StructField(\"geom\", GeometryUDT) ::\n+      Nil)\n+\n+  val BareElementEncoder: Encoder[Row] = RowEncoder(BareElementSchema)\n+\n+  val TaggedVersionedElementSchema = StructType(\n+    StructField(\"changeset\", LongType, nullable = false) ::\n+      StructField(\"id\", LongType, nullable = false) ::\n+      StructField(\"tags\", MapType(StringType, StringType, valueContainsNull = false), nullable = false) ::\n+      StructField(\"version\", IntegerType, nullable = false) ::\n+      StructField(\"minorVersion\", IntegerType, nullable = false) ::\n+      StructField(\"updated\", TimestampType, nullable = false) ::\n+      StructField(\"validUntil\", TimestampType) ::\n+      StructField(\"geom\", GeometryUDT) ::\n+      Nil)\n+\n+  val TaggedVersionedElementEncoder: Encoder[Row] = RowEncoder(TaggedVersionedElementSchema)\n+\n+  val VersionedElementSchema = StructType(\n+    StructField(\"changeset\", LongType, nullable = false) ::\n+      StructField(\"id\", LongType, nullable = false) ::\n+      StructField(\"version\", IntegerType, nullable = false) ::\n+      StructField(\"minorVersion\", IntegerType, nullable = false) ::\n+      StructField(\"updated\", TimestampType, nullable = false) ::\n+      StructField(\"validUntil\", TimestampType) ::\n+      StructField(\"geom\", GeometryUDT) ::\n+      Nil)\n+\n+  val VersionedElementEncoder: Encoder[Row] = RowEncoder(VersionedElementSchema)\n+\n+  /**\n+    * Snapshot pre-processed elements.\n+    *\n+    * @param df        Elements (including 'validUntil column)\n+    * @param timestamp Optional timestamp to snapshot at\n+    * @return DataFrame containing valid elements at timestamp (or now)\n+    */\n+  def snapshot(df: DataFrame, timestamp: Timestamp = null): DataFrame = {\n+    import df.sparkSession.implicits._\n+\n+    df\n+      .where(\n+        'updated <= coalesce(lit(timestamp), current_timestamp)\n+          and coalesce(lit(timestamp), current_timestamp) < coalesce('validUntil, date_add(current_timestamp, 1)))\n+  }\n+\n+  /**\n+    * Pre-process nodes. Copies coordinates + tags from versions prior to being deleted (!'visible), as they're cleared\n+    * out otherwise, and adds 'validUntil based on the creation timestamp of the next version. Nodes with null\n+    * 'validUntil values are currently valid.\n+    *\n+    * @param history DataFrame containing nodes.\n+    * @return processed nodes.\n+    */\n+  def preprocessNodes(history: DataFrame, extent: Option[Extent] = None): DataFrame = {\n+    import history.sparkSession.implicits._\n+\n+    val filteredHistory =\n+      extent match {\n+        case Some(e) =>\n+          history\n+            .where('lat > e.ymin and 'lat < e.ymax)\n+            .where('lon > e.xmin and 'lon < e.xmax)\n+        case None =>\n+          history\n+      }\n+\n+    if (filteredHistory.columns.contains(\"validUntil\")) {\n+      filteredHistory\n+    } else {\n+      @transient val idByVersion = Window.partitionBy('id).orderBy('version)\n+\n+      // when a node has been deleted, it doesn't include any tags; use a window function to retrieve the last tags\n+      // present and use those\n+      filteredHistory\n+        .where('type === \"node\")\n+        .repartition('id)\n+        .select(\n+          'id,\n+          when(!'visible and (lag('tags, 1) over idByVersion).isNotNull,\n+            lag('tags, 1) over idByVersion)\n+          .otherwise('tags) as 'tags,\n+          when(!'visible, lit(Float.NaN)).otherwise(asFloat('lat)) as 'lat,\n+          when(!'visible, lit(Float.NaN)).otherwise(asFloat('lon)) as 'lon,\n+          'changeset,\n+          'timestamp,\n+          (lead('timestamp, 1) over idByVersion) as 'validUntil,\n+          'uid,\n+          'user,\n+          'version,\n+          'visible,\n+          !((lag('lat, 1) over idByVersion) <=> 'lat and (lag('lon, 1) over idByVersion) <=> 'lon) as 'geometryChanged\n+        )\n+    }\n+  }\n+\n+  /**\n+    * Pre-process ways. Copies tags from versions prior to being deleted (!'visible), as they're cleared out\n+    * otherwise, dereferences 'nds.ref, and adds 'validUntil based on the creation timestamp of the next version. Ways\n+    * with null 'validUntil values are currently valid.\n+    *\n+    * @param history DataFrame containing ways.\n+    * @return processed ways.\n+    */\n+  def preprocessWays(history: DataFrame): DataFrame = {\n+    import history.sparkSession.implicits._\n+\n+    if (history.columns.contains(\"validUntil\")) {\n+      history\n+    } else {\n+      @transient val idByVersion = Window.partitionBy('id).orderBy('version)\n+\n+      // when a node has been deleted, it doesn't include any tags; use a window function to retrieve the last tags\n+      // present and use those\n+      history\n+        .where('type === \"way\")\n+        .repartition('id)\n+        .select(\n+          'id,\n+          when(!'visible and (lag('tags, 1) over idByVersion).isNotNull,\n+            lag('tags, 1) over idByVersion)\n+          .otherwise('tags) as 'tags,\n+          $\"nds.ref\" as 'nds,\n+          'changeset,\n+          'timestamp,\n+          (lead('timestamp, 1) over idByVersion) as 'validUntil,\n+          'uid,\n+          'user,\n+          'version,\n+          'visible,\n+          !((lag('nds, 1) over idByVersion) <=> 'nds) as 'geometryChanged\n+        )\n+    }\n+  }\n+\n+  /**\n+    * Pre-process relations. Copies tags from versions prior to being deleted (!'visible), as they're cleared out\n+    * otherwise, and adds 'validUntil based on the creation timestamp of the next version. Relations with null\n+    * 'validUntil values are currently valid.\n+    *\n+    * @param history DataFrame containing relations.\n+    * @return processed relations.\n+    */\n+  def preprocessRelations(history: DataFrame): DataFrame = {\n+    import history.sparkSession.implicits._\n+\n+    if (history.columns.contains(\"validUntil\")) {\n+      history\n+    } else {\n+      @transient val idByUpdated = Window.partitionBy('id).orderBy('version)\n+\n+      // when an element has been deleted, it doesn't include any tags; use a window function to retrieve the last tags\n+      // present and use those\n+      history\n+        .where('type === \"relation\")\n+        .repartition('id)\n+        .select(\n+          'id,\n+          when(!'visible and (lag('tags, 1) over idByUpdated).isNotNull,\n+            lag('tags, 1) over idByUpdated)\n+          .otherwise('tags) as 'tags,\n+          compressMemberTypes('members) as 'members,\n+          'changeset,\n+          'timestamp,\n+          (lead('timestamp, 1) over idByUpdated) as 'validUntil,\n+          'uid,\n+          'user,\n+          'version,\n+          'visible)\n+    }\n+  }\n+\n+  /**\n+    * Construct all geometries for a set of elements.\n+    *\n+    * @param elements DataFrame containing node, way, and relation elements\n+    * @return DataFrame containing geometries.\n+    */\n+  def constructGeometries(elements: DataFrame): DataFrame = {\n+    import elements.sparkSession.implicits._\n+\n+    val nodes = ProcessOSM.preprocessNodes(elements)\n+\n+    val nodeGeoms = ProcessOSM.constructPointGeometries(nodes)\n+      .withColumn(\"minorVersion\", lit(0))\n+\n+    val wayGeoms = ProcessOSM.reconstructWayGeometries(elements, nodes)\n+\n+    val relationGeoms = ProcessOSM.reconstructRelationGeometries(elements, wayGeoms)\n+\n+    nodeGeoms\n+      .union(wayGeoms.where(size('tags) > 0).drop('geometryChanged))\n+      .union(relationGeoms)\n+  }\n+\n+  /**\n+    * Construct point geometries. \"Uninteresting\" nodes are not included, so this is not suitable for way/relation\n+    * assembly.\n+    *\n+    * @param nodes DataFrame containing nodes\n+    * @return Nodes as Point geometries\n+    */\n+  def constructPointGeometries(nodes: DataFrame): DataFrame = {\n+    import nodes.sparkSession.implicits._\n+\n+    val ns = preprocessNodes(nodes)\n+      .where(size('tags) > 0)\n+\n+    ns\n+      // fetch the last version of a node within a single changeset\n+      .select('changeset, 'id, 'version, 'timestamp)\n+      .groupBy('changeset, 'id)\n+      .agg(max('version).cast(IntegerType) as 'version, max('timestamp) as 'updated)\n+      .join(ns.drop('changeset), Seq(\"id\", \"version\"))\n+      .select(\n+        lit(NodeType) as '_type,\n+        'id,\n+        st_makePoint('lon, 'lat) as 'geom,\n+        'tags,\n+        'changeset,\n+        'updated,\n+        'validUntil,\n+        'visible,\n+        'version)\n+  }\n+\n+  /**\n+    * Reconstruct way geometries.\n+    *\n+    * Nodes and ways contain implicit timestamps that will be used to generate minor versions of geometry that they're\n+    * associated with (each entry that exists within a changeset).\n+    *\n+    * @param _ways        DataFrame containing ways to reconstruct.\n+    * @param _nodes       DataFrame containing nodes used to construct ways.\n+    * @param _nodesToWays Optional lookup table.\n+    * @return Way geometries.\n+    */\n+  def reconstructWayGeometries(_ways: DataFrame, _nodes: DataFrame, _nodesToWays: Option[DataFrame] = None)(implicit cache: Caching = Caching.none, cachePartitions: Option[Int] = None): DataFrame = {\n+    implicit val ss: SparkSession = _ways.sparkSession\n+    import ss.implicits._\n+\n+    @transient val idByVersion = Window.partitionBy('id).orderBy('version)\n+\n+    val nodes = preprocessNodes(_nodes)\n+      // no longer correct after filtering out unchanged geometries\n+      .drop('validUntil)\n+      .where('geometryChanged)\n+      .drop('geometryChanged)\n+      // re-calculate validUntil windows\n+      .withColumn(\"validUntil\", lead('timestamp, 1) over idByVersion)\n+\n+    val ways = preprocessWays(_ways)\n+      .withColumn(\"isArea\", isArea('tags))\n+\n+    // Create (or re-use) a lookup table for node → ways\n+    val nodesToWays = _nodesToWays.getOrElse[DataFrame] {\n+      ways\n+        .select(explode('nds) as 'id, 'id as 'wayId, 'version, 'timestamp, 'validUntil)\n+    }\n+\n+    // Create a way entry for each changeset in which a node was modified, containing the timestamp of the node that\n+    // triggered the association. This will later be used to assemble ways at each of those points in time. If you need\n+    // authorship, join on changesets\n+    val waysByChangeset = nodes\n+      .select('changeset, 'id, 'timestamp as 'updated)\n+      .join(nodesToWays, Array(\"id\"))\n+      .where('timestamp <= 'updated and 'updated < coalesce('validUntil, current_timestamp))\n+      .select('changeset, 'wayId as 'id, 'version, 'updated)\n+\n+    val allWayVersions = waysByChangeset\n+      // Union with raw ways to include those in the time line (if they weren't already triggered by node modifications\n+      // at the same time)\n+      .union(ways.select('changeset, 'id, 'version, 'timestamp as 'updated))\n+      // If a node and a way were modified within the same changeset at different times, there will be multiple entries\n+      // per changeset (with different timestamps). There should only be one per changeset.\n+      .groupBy('changeset, 'id)\n+      .agg(max('version).cast(IntegerType) as 'version, max('updated) as 'updated)\n+      .join(ways.select('id, 'version, 'nds, 'isArea), Seq(\"id\", \"version\"))\n+\n+    val explodedWays = allWayVersions\n+      .select('changeset, 'id, 'version, 'updated, 'isArea, posexplode_outer('nds) as Array(\"idx\", \"ref\"))\n+      // repartition including updated timestamp to avoid skew (version is insufficient, as\n+      // multiple instances may exist with the same version)\n+      .repartition('id, 'updated)\n+\n+    val waysAndNodes = explodedWays\n+      .join(nodes.select('id as 'ref, 'timestamp, 'validUntil, 'lat, 'lon), Seq(\"ref\"), \"left_outer\")\n+      .where('timestamp <= 'updated and 'updated < coalesce('validUntil, current_timestamp))\n+\n+    implicit val encoder: Encoder[Row] = BareElementEncoder\n+\n+    val wayGeoms = waysAndNodes\n+      .select('changeset, 'id, 'version, 'updated, 'isArea, 'idx, 'lat, 'lon)\n+      .repartition('id, 'updated)\n+      .sortWithinPartitions('id, 'version, 'updated, 'idx)\n+      .drop('idx)\n+      // leverage partitioning (avoids repeated (de-)serialization of merged coordinate arrays) instead of using a UDAF\n+      .mapPartitions(rows => {\n+        rows\n+          .toVector\n+          .groupBy(row =>\n+            (row.getAs[Long](\"changeset\"), row.getAs[Long](\"id\"), row.getAs[Integer](\"version\"), row.getAs[Timestamp](\"updated\"))\n+          )\n+          .map {\n+            case ((changeset, id, version, updated), rows: Seq[Row]) =>\n+              val isArea = rows.head.getAs[Boolean](\"isArea\")\n+              val geom = rows.map(row =>\n+                Seq(Option(row.get(row.fieldIndex(\"lon\"))).map(_.asInstanceOf[Float]).getOrElse(Float.NaN),\n+                  Option(row.get(row.fieldIndex(\"lat\"))).map(_.asInstanceOf[Float]).getOrElse(Float.NaN))) match {\n+                    // no coordinates provided\n+                    case coords if coords.isEmpty => Some(GeomFactory.factory.createLineString(Array[Coordinate]()))\n+                    // some of the coordinates are empty; this is invalid\n+                    case coords if coords.exists(Option(_).isEmpty) => None\n+                    // some of the coordinates are invalid\n+                    case coords if coords.exists(_.exists(_.isNaN)) => None\n+                    // 1 pair of coordinates provided\n+                    case coords if coords.length == 1 =>\n+                      Some(GeomFactory.factory.createPoint(new Coordinate(coords.head.head, coords.head.last)))\n+                    case coords => {\n+                      val coordinates = coords.map { xy => new Coordinate(xy.head, xy.last) }.toArray\n+                      val line = GeomFactory.factory.createLineString(coordinates)\n+\n+                      if (isArea && line.getNumPoints >= 4 && line.isClosed)\n+                        Some(GeomFactory.factory.createPolygon(line.getCoordinateSequence()))\n+                      else\n+                        Some(line)\n+                    }\n+                  }\n+\n+              val wkb = geom match {\n+                case Some(g) if g.isValid => g\n+                case _ => null\n+              }\n+\n+              new GenericRowWithSchema(Array(changeset, id, version, updated, wkb), BareElementSchema): Row\n+          }\n+          .toIterator\n+      })\n+\n+    @transient val idAndVersionByUpdated = Window.partitionBy('id, 'version).orderBy('updated)\n+    @transient val idByUpdated = Window.partitionBy('id).orderBy('updated)\n+\n+    wayGeoms\n+      // Assign `minorVersion` and rewrite `validUntil` to match\n+      .withColumn(\"validUntil\", lead('updated, 1) over idByUpdated)\n+      .withColumn(\"minorVersion\", (row_number over idAndVersionByUpdated) - 1)\n+      .withColumn(\"geometryChanged\", !((lag('geom, 1) over idByUpdated) <=> 'geom))\n+      .join(ways.select('id, 'version, 'tags, 'visible), Seq(\"id\", \"version\"))\n+      .select(\n+        lit(WayType) as '_type,\n+        'id,\n+        'geom,\n+        'tags,\n+        'changeset,\n+        'updated,\n+        'validUntil,\n+        'visible,\n+        'version,\n+        'minorVersion,\n+        'geometryChanged)\n+  }\n+\n+  private def getRelationMembers(relations: DataFrame, geoms: DataFrame) = {\n+    implicit val ss: SparkSession = relations.sparkSession\n+    import ss.implicits._\n+\n+    // way to relation lookup table (missing 'role, since it's not needed here)\n+    val waysToRelations = relations\n+      .select(explode('members) as 'member, 'id as 'relationId, 'version, 'timestamp, 'validUntil)\n+      .withColumn(\"type\", $\"member.type\")\n+      .withColumn(\"id\", $\"member.ref\")\n+      .drop('member)\n+\n+    @transient val idByVersion = Window.partitionBy('id).orderBy('version)\n+\n+    // Create a relation entry for each changeset in which a geometry was modified, containing the timestamp and\n+    // changeset of the geometry that triggered the association. This will later be used to assemble relations at each\n+    // of those points in time.\n+    // If you need authorship, join on changesets\n+    val relationsByChangeset = geoms\n+      .where('geometryChanged)\n+      .drop('validUntil)\n+      // re-calculate validUntil windows\n+      .withColumn(\"validUntil\", lead('updated, 1) over idByVersion)\n+      // TODO when expanding beyond relations referring to ways, geoms should include 'type for the join to work properly\n+      .withColumn(\"type\", lit(WayType))\n+      .select('type, 'changeset, 'id, 'updated)\n+      .join(waysToRelations, Seq(\"id\", \"type\"))\n+      .where(waysToRelations(\"timestamp\") <= geoms(\"updated\") and\n+        geoms(\"updated\") < coalesce(waysToRelations(\"validUntil\"), current_timestamp))\n+      .select('changeset, 'relationId as 'id, 'version, 'updated)\n+\n+    @transient val idAndVersionByUpdated = Window.partitionBy('id, 'version).orderBy('updated)\n+    @transient val idByUpdated = Window.partitionBy('id).orderBy('updated)\n+\n+    val allRelationVersions = relationsByChangeset\n+      // Union with raw relations to include those in the time line (if they weren't already triggered by geometry\n+      // modifications at the same time)\n+      .union(relations.select('changeset, 'id, 'version, 'timestamp as 'updated))\n+      // If a node, a way, and/or a relation were modified within the same changeset at different times, there will be\n+      // multiple entries with different timestamps; this reduces them down to a single update per changeset.\n+      .groupBy('changeset, 'id)\n+      .agg(max('version).cast(IntegerType) as 'version, max('updated) as 'updated)\n+      .join(relations.select('id, 'version, 'members), Seq(\"id\", \"version\"))\n+      // assign `minorVersion` and rewrite `validUntil` to match\n+      // this is done early (at the expense of passing through the shuffle w/ exploded 'members) to avoid extremely\n+      // large partitions (see: Germany w/ 7k+ geometry versions) after geometries have been constructed\n+      .withColumn(\"validUntil\", lead('updated, 1) over idByUpdated)\n+      .withColumn(\"minorVersion\", (row_number over idAndVersionByUpdated) - 1)\n+\n+    allRelationVersions\n+      .select('changeset, 'id, 'version, 'minorVersion, 'updated, 'validUntil, explode_outer('members) as \"member\")\n+      .select(\n+        'changeset,\n+        'id,\n+        'version,\n+        'minorVersion,\n+        'updated,\n+        'validUntil,\n+        'member.getField(\"type\") as 'type,\n+        'member.getField(\"ref\") as 'ref,\n+        'member.getField(\"role\") as 'role\n+      )\n+      .distinct\n+  }\n+\n+  /**\n+    * Reconstruct relation geometries.\n+    *\n+    * Nodes and ways contain implicit timestamps that will be used to generate minor versions of geometry that they're\n+    * associated with (each entry that exists within a changeset).\n+    *\n+    * @param _relations DataFrame containing relations to reconstruct.\n+    * @param geoms      DataFrame containing way geometries to use in reconstruction.\n+    * @return Relations geometries.\n+    */\n+  def reconstructRelationGeometries(\n+    _relations: DataFrame,\n+    geoms: DataFrame\n+  )(implicit cache: Caching = Caching.none, cachePartitions: Option[Int] = None): DataFrame = {\n+    val relations = preprocessRelations(_relations)\n+\n+    reconstructMultiPolygonRelationGeometries(relations, geoms)\n+      .union(reconstructRouteRelationGeometries(relations, geoms))\n+  }\n+\n+  def reconstructMultiPolygonRelationGeometries(\n+    _relations: DataFrame,\n+    geoms: DataFrame\n+  )(implicit cache: Caching = Caching.none, cachePartitions: Option[Int] = None): DataFrame = {\n+    implicit val ss: SparkSession = _relations.sparkSession\n+    import ss.implicits._\n+\n+    val relations = preprocessRelations(_relations)\n+      .where(isMultiPolygon('tags))\n+\n+    val members = getRelationMembers(relations, geoms)\n+      .where('role.isin(MultiPolygonRoles: _*))\n+      // TODO when expanding beyond multipolygons, geoms should include 'type for the join to work properly\n+      .join(\n+        geoms.select(\n+          lit(WayType) as 'type,\n+          'id as \"ref\",\n+          'updated as 'memberUpdated,\n+          'validUntil as 'memberValidUntil,\n+          'geom), Seq(\"type\", \"ref\"), \"left_outer\")\n+      .where(\n+        ('memberUpdated.isNull and 'memberValidUntil.isNull and 'geom.isNull) or // allow left outer join artifacts through\n+          ('memberUpdated <= 'updated and 'updated < coalesce('memberValidUntil, current_timestamp)))\n+      .drop('memberUpdated)\n+      .drop('memberValidUntil)\n+      .drop('ref)\n+\n+    implicit val encoder: Encoder[Row] = VersionedElementEncoder\n+\n+    // leverage partitioning (avoids repeated (de-)serialization of merged coordinate arrays)\n+    val relationGeoms = members\n+      .repartition('changeset, 'id, 'version, 'minorVersion, 'updated, 'validUntil)\n+      .mapPartitions(rows => {\n+        rows\n+          .toVector\n+          .groupBy(row =>\n+            (row.getAs[Long](\"changeset\"), row.getAs[Long](\"id\"), row.getAs[Integer](\"version\"), row.getAs[Integer](\"minorVersion\"), row.getAs[Timestamp](\"updated\"), row.getAs[Timestamp](\"validUntil\"))\n+          )\n+          .map {\n+            case ((changeset, id, version, minorVersion, updated, validUntil), rows: Seq[Row]) =>\n+              val types = rows.map(_.getAs[Byte](\"type\"))\n+              val roles = rows.map(_.getAs[String](\"role\"))\n+              val geoms = rows.map(_.getAs[Geometry](\"geom\"))\n+\n+              val wkb = buildMultiPolygon(id, version, updated, types, roles, geoms).orNull\n+\n+              new GenericRowWithSchema(Array(changeset, id, version, minorVersion, updated, validUntil, wkb), VersionedElementSchema): Row\n+          }\n+          .toIterator\n+      })\n+\n+    // Join metadata to avoid passing it through exploded shuffles\n+    relationGeoms\n+      .join(relations.select('id, 'version, 'tags, 'visible), Array(\"id\", \"version\"))\n+      .select(\n+        lit(RelationType) as '_type,\n+        'id,\n+        'geom,\n+        'tags,\n+        'changeset,\n+        'updated,\n+        'validUntil,\n+        'visible,\n+        'version,\n+        'minorVersion)\n+  }\n+\n+  def reconstructRouteRelationGeometries(\n+    _relations: DataFrame,\n+    geoms: DataFrame\n+  )(implicit cache: Caching = Caching.none, cachePartitions: Option[Int] = None): DataFrame = {\n+    implicit val ss: SparkSession = _relations.sparkSession\n+    import ss.implicits._\n+\n+    val relations = preprocessRelations(_relations)\n+      .where(isRoute('tags))\n+\n+    val members = getRelationMembers(relations, geoms)\n+      // TODO when expanding beyond way-based routes, geoms should include 'type for the join to work properly\n+      .join(\n+        geoms.select(\n+          lit(WayType) as 'type,\n+          'id as \"ref\",\n+          'updated as 'memberUpdated,\n+          'validUntil as 'memberValidUntil,\n+          'geom), Seq(\"type\", \"ref\"), \"left_outer\")\n+      .where(\n+        ('memberUpdated.isNull and 'memberValidUntil.isNull and 'geom.isNull) or // allow left outer join artifacts through\n+          ('memberUpdated <= 'updated and 'updated < coalesce('memberValidUntil, current_timestamp)))\n+      .drop('memberUpdated)\n+      .drop('memberValidUntil)\n+      .drop('ref)\n+\n+    implicit val encoder: Encoder[Row] = TaggedVersionedElementEncoder\n+\n+    // leverage partitioning (avoids repeated (de-)serialization of merged coordinate arrays)\n+    val relationGeoms = members\n+      .repartition('changeset, 'id, 'version, 'minorVersion, 'updated, 'validUntil)\n+      .mapPartitions(rows => {\n+        rows\n+          .toVector\n+          .groupBy(row =>\n+            (row.getAs[Long](\"changeset\"), row.getAs[Long](\"id\"), row.getAs[Integer](\"version\"), row.getAs[Integer](\"minorVersion\"), row.getAs[Timestamp](\"updated\"), row.getAs[Timestamp](\"validUntil\"))\n+          )\n+          .flatMap {\n+            case ((changeset, id, version, minorVersion, updated, validUntil), rows: Seq[Row]) =>\n+              val types = rows.map(_.getAs[Byte](\"type\"))\n+              val roles = rows.map(_.getAs[String](\"role\"))\n+              val geoms = rows.map(_.getAs[Geometry](\"geom\"))\n+\n+              buildRoute(id, version, updated, types, roles, geoms) match {\n+                case Some(components) =>\n+                  components.map {\n+                    case (\"\", wkb) =>\n+                      // no role\n+                      new GenericRowWithSchema(Array(changeset, id, Map(), version, minorVersion, updated, validUntil, wkb), TaggedVersionedElementSchema): Row\n+                    case (role, wkb) =>\n+                      new GenericRowWithSchema(Array(changeset, id, Map(\"role\" -> role), version, minorVersion, updated, validUntil, wkb), TaggedVersionedElementSchema): Row\n+                  }\n+                case None =>\n+                  // no geometry\n+                  Seq(new GenericRowWithSchema(Array(changeset, id, Map(), version, minorVersion, updated, validUntil, null), TaggedVersionedElementSchema): Row)\n+              }\n+          }\n+          .toIterator\n+      })\n+\n+    // Join metadata to avoid passing it through exploded shuffles\n+    relationGeoms\n+      .join(relations.select('id, 'version, 'tags as 'originalTags, 'visible), Seq(\"id\", \"version\"))\n+      .select(\n+        lit(RelationType) as '_type,\n+        'id,\n+        'geom,\n+        mergeTags('originalTags, 'tags) as 'tags,\n+        'changeset,\n+        'updated,\n+        'validUntil,\n+        'visible,\n+        'version,\n+        'minorVersion)\n+  }\n+\n+  /**\n+    * Augment geometries with user metadata.\n+    *\n+    * @param geoms      Geometries to augment.\n+    * @param changesets Changesets DataFrame with user metadata.\n+    * @return Geometries augmented with user metadata.\n+    */\n+  def addUserMetadata(geoms: DataFrame, changesets: DataFrame): DataFrame = {\n+    import geoms.sparkSession.implicits._\n+\n+    geoms\n+      .join(changesets.select('id as 'changeset, 'uid, 'user), Array(\"changeset\"))\n+  }\n+\n+  /*\n+  def geometriesByRegion(nodeGeoms: DataFrame, wayGeoms: DataFrame): DataFrame = {","path":"src/main/scala/vectorpipe/osm/internal/ProcessOSM.scala","position":644,"original_position":644,"commit_id":"c9f8e4a836ed8c8f5bc96eb78d9f1914fc7ae74f","original_commit_id":"c9f8e4a836ed8c8f5bc96eb78d9f1914fc7ae74f","user":{"login":"jbouffard","id":11041248,"avatar_url":"https://avatars3.githubusercontent.com/u/11041248?v=4","gravatar_id":"","url":"https://api.github.com/users/jbouffard","html_url":"https://github.com/jbouffard","followers_url":"https://api.github.com/users/jbouffard/followers","following_url":"https://api.github.com/users/jbouffard/following{/other_user}","gists_url":"https://api.github.com/users/jbouffard/gists{/gist_id}","starred_url":"https://api.github.com/users/jbouffard/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/jbouffard/subscriptions","organizations_url":"https://api.github.com/users/jbouffard/orgs","repos_url":"https://api.github.com/users/jbouffard/repos","events_url":"https://api.github.com/users/jbouffard/events{/privacy}","received_events_url":"https://api.github.com/users/jbouffard/received_events","type":"User","site_admin":false},"body":"I commented this out because I wasn't sure if this was a feature we needed right now and because I have a few questions on it. I can re-implement it though if it's desired for this PR.","created_at":"2018-05-30T12:59:32Z","updated_at":"2018-05-30T12:59:32Z","html_url":"https://github.com/geotrellis/vectorpipe/pull/54#discussion_r191756652","pull_request_url":"https://api.github.com/repos/geotrellis/vectorpipe/pulls/54","author_association":"NONE","_links":{"self":{"href":"https://api.github.com/repos/geotrellis/vectorpipe/pulls/comments/191756652"},"html":{"href":"https://github.com/geotrellis/vectorpipe/pull/54#discussion_r191756652"},"pull_request":{"href":"https://api.github.com/repos/geotrellis/vectorpipe/pulls/54"}}},"pull_request":{"url":"https://api.github.com/repos/geotrellis/vectorpipe/pulls/54","id":191211151,"html_url":"https://github.com/geotrellis/vectorpipe/pull/54","diff_url":"https://github.com/geotrellis/vectorpipe/pull/54.diff","patch_url":"https://github.com/geotrellis/vectorpipe/pull/54.patch","issue_url":"https://api.github.com/repos/geotrellis/vectorpipe/issues/54","number":54,"state":"open","locked":false,"title":"[WIP] osmesa Code Migration","user":{"login":"jbouffard","id":11041248,"avatar_url":"https://avatars3.githubusercontent.com/u/11041248?v=4","gravatar_id":"","url":"https://api.github.com/users/jbouffard","html_url":"https://github.com/jbouffard","followers_url":"https://api.github.com/users/jbouffard/followers","following_url":"https://api.github.com/users/jbouffard/following{/other_user}","gists_url":"https://api.github.com/users/jbouffard/gists{/gist_id}","starred_url":"https://api.github.com/users/jbouffard/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/jbouffard/subscriptions","organizations_url":"https://api.github.com/users/jbouffard/orgs","repos_url":"https://api.github.com/users/jbouffard/repos","events_url":"https://api.github.com/users/jbouffard/events{/privacy}","received_events_url":"https://api.github.com/users/jbouffard/received_events","type":"User","site_admin":false},"body":"This PR seeks to migrate the code dealing with creating JTS DataFrames from osmesa to vectorpipe. By doing this, DataFrame creating will be more performant and easier. In addition to the code migration, new API will developed so that the new logic can be accessed in an easy way.","created_at":"2018-05-29T17:01:02Z","updated_at":"2018-05-30T12:59:32Z","closed_at":null,"merged_at":null,"merge_commit_sha":"8d1f50577290c7470f8ecf93946daebe19c80c3f","assignee":null,"assignees":[],"requested_reviewers":[],"requested_teams":[],"labels":[],"milestone":null,"commits_url":"https://api.github.com/repos/geotrellis/vectorpipe/pulls/54/commits","review_comments_url":"https://api.github.com/repos/geotrellis/vectorpipe/pulls/54/comments","review_comment_url":"https://api.github.com/repos/geotrellis/vectorpipe/pulls/comments{/number}","comments_url":"https://api.github.com/repos/geotrellis/vectorpipe/issues/54/comments","statuses_url":"https://api.github.com/repos/geotrellis/vectorpipe/statuses/c9f8e4a836ed8c8f5bc96eb78d9f1914fc7ae74f","head":{"label":"jbouffard:feature/osmesa-code-migration","ref":"feature/osmesa-code-migration","sha":"c9f8e4a836ed8c8f5bc96eb78d9f1914fc7ae74f","user":{"login":"jbouffard","id":11041248,"avatar_url":"https://avatars3.githubusercontent.com/u/11041248?v=4","gravatar_id":"","url":"https://api.github.com/users/jbouffard","html_url":"https://github.com/jbouffard","followers_url":"https://api.github.com/users/jbouffard/followers","following_url":"https://api.github.com/users/jbouffard/following{/other_user}","gists_url":"https://api.github.com/users/jbouffard/gists{/gist_id}","starred_url":"https://api.github.com/users/jbouffard/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/jbouffard/subscriptions","organizations_url":"https://api.github.com/users/jbouffard/orgs","repos_url":"https://api.github.com/users/jbouffard/repos","events_url":"https://api.github.com/users/jbouffard/events{/privacy}","received_events_url":"https://api.github.com/users/jbouffard/received_events","type":"User","site_admin":false},"repo":{"id":116857275,"name":"vectorpipe","full_name":"jbouffard/vectorpipe","owner":{"login":"jbouffard","id":11041248,"avatar_url":"https://avatars3.githubusercontent.com/u/11041248?v=4","gravatar_id":"","url":"https://api.github.com/users/jbouffard","html_url":"https://github.com/jbouffard","followers_url":"https://api.github.com/users/jbouffard/followers","following_url":"https://api.github.com/users/jbouffard/following{/other_user}","gists_url":"https://api.github.com/users/jbouffard/gists{/gist_id}","starred_url":"https://api.github.com/users/jbouffard/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/jbouffard/subscriptions","organizations_url":"https://api.github.com/users/jbouffard/orgs","repos_url":"https://api.github.com/users/jbouffard/repos","events_url":"https://api.github.com/users/jbouffard/events{/privacy}","received_events_url":"https://api.github.com/users/jbouffard/received_events","type":"User","site_admin":false},"private":false,"html_url":"https://github.com/jbouffard/vectorpipe","description":"Convert Vector data to VectorTiles with GeoTrellis.","fork":true,"url":"https://api.github.com/repos/jbouffard/vectorpipe","forks_url":"https://api.github.com/repos/jbouffard/vectorpipe/forks","keys_url":"https://api.github.com/repos/jbouffard/vectorpipe/keys{/key_id}","collaborators_url":"https://api.github.com/repos/jbouffard/vectorpipe/collaborators{/collaborator}","teams_url":"https://api.github.com/repos/jbouffard/vectorpipe/teams","hooks_url":"https://api.github.com/repos/jbouffard/vectorpipe/hooks","issue_events_url":"https://api.github.com/repos/jbouffard/vectorpipe/issues/events{/number}","events_url":"https://api.github.com/repos/jbouffard/vectorpipe/events","assignees_url":"https://api.github.com/repos/jbouffard/vectorpipe/assignees{/user}","branches_url":"https://api.github.com/repos/jbouffard/vectorpipe/branches{/branch}","tags_url":"https://api.github.com/repos/jbouffard/vectorpipe/tags","blobs_url":"https://api.github.com/repos/jbouffard/vectorpipe/git/blobs{/sha}","git_tags_url":"https://api.github.com/repos/jbouffard/vectorpipe/git/tags{/sha}","git_refs_url":"https://api.github.com/repos/jbouffard/vectorpipe/git/refs{/sha}","trees_url":"https://api.github.com/repos/jbouffard/vectorpipe/git/trees{/sha}","statuses_url":"https://api.github.com/repos/jbouffard/vectorpipe/statuses/{sha}","languages_url":"https://api.github.com/repos/jbouffard/vectorpipe/languages","stargazers_url":"https://api.github.com/repos/jbouffard/vectorpipe/stargazers","contributors_url":"https://api.github.com/repos/jbouffard/vectorpipe/contributors","subscribers_url":"https://api.github.com/repos/jbouffard/vectorpipe/subscribers","subscription_url":"https://api.github.com/repos/jbouffard/vectorpipe/subscription","commits_url":"https://api.github.com/repos/jbouffard/vectorpipe/commits{/sha}","git_commits_url":"https://api.github.com/repos/jbouffard/vectorpipe/git/commits{/sha}","comments_url":"https://api.github.com/repos/jbouffard/vectorpipe/comments{/number}","issue_comment_url":"https://api.github.com/repos/jbouffard/vectorpipe/issues/comments{/number}","contents_url":"https://api.github.com/repos/jbouffard/vectorpipe/contents/{+path}","compare_url":"https://api.github.com/repos/jbouffard/vectorpipe/compare/{base}...{head}","merges_url":"https://api.github.com/repos/jbouffard/vectorpipe/merges","archive_url":"https://api.github.com/repos/jbouffard/vectorpipe/{archive_format}{/ref}","downloads_url":"https://api.github.com/repos/jbouffard/vectorpipe/downloads","issues_url":"https://api.github.com/repos/jbouffard/vectorpipe/issues{/number}","pulls_url":"https://api.github.com/repos/jbouffard/vectorpipe/pulls{/number}","milestones_url":"https://api.github.com/repos/jbouffard/vectorpipe/milestones{/number}","notifications_url":"https://api.github.com/repos/jbouffard/vectorpipe/notifications{?since,all,participating}","labels_url":"https://api.github.com/repos/jbouffard/vectorpipe/labels{/name}","releases_url":"https://api.github.com/repos/jbouffard/vectorpipe/releases{/id}","deployments_url":"https://api.github.com/repos/jbouffard/vectorpipe/deployments","created_at":"2018-01-09T18:57:09Z","updated_at":"2018-01-09T18:57:11Z","pushed_at":"2018-05-30T12:45:50Z","git_url":"git://github.com/jbouffard/vectorpipe.git","ssh_url":"git@github.com:jbouffard/vectorpipe.git","clone_url":"https://github.com/jbouffard/vectorpipe.git","svn_url":"https://github.com/jbouffard/vectorpipe","homepage":"https://geotrellis.github.io/vectorpipe/","size":10390,"stargazers_count":0,"watchers_count":0,"language":"Scala","has_issues":false,"has_projects":true,"has_downloads":true,"has_wiki":false,"has_pages":false,"forks_count":0,"mirror_url":null,"archived":false,"open_issues_count":0,"license":{"key":"other","name":"Other","spdx_id":null,"url":null},"forks":0,"open_issues":0,"watchers":0,"default_branch":"master"}},"base":{"label":"geotrellis:master","ref":"master","sha":"e71f8d202633c433963d8792a6c3f744c7c7267b","user":{"login":"geotrellis","id":3129444,"avatar_url":"https://avatars0.githubusercontent.com/u/3129444?v=4","gravatar_id":"","url":"https://api.github.com/users/geotrellis","html_url":"https://github.com/geotrellis","followers_url":"https://api.github.com/users/geotrellis/followers","following_url":"https://api.github.com/users/geotrellis/following{/other_user}","gists_url":"https://api.github.com/users/geotrellis/gists{/gist_id}","starred_url":"https://api.github.com/users/geotrellis/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/geotrellis/subscriptions","organizations_url":"https://api.github.com/users/geotrellis/orgs","repos_url":"https://api.github.com/users/geotrellis/repos","events_url":"https://api.github.com/users/geotrellis/events{/privacy}","received_events_url":"https://api.github.com/users/geotrellis/received_events","type":"Organization","site_admin":false},"repo":{"id":71492746,"name":"vectorpipe","full_name":"geotrellis/vectorpipe","owner":{"login":"geotrellis","id":3129444,"avatar_url":"https://avatars0.githubusercontent.com/u/3129444?v=4","gravatar_id":"","url":"https://api.github.com/users/geotrellis","html_url":"https://github.com/geotrellis","followers_url":"https://api.github.com/users/geotrellis/followers","following_url":"https://api.github.com/users/geotrellis/following{/other_user}","gists_url":"https://api.github.com/users/geotrellis/gists{/gist_id}","starred_url":"https://api.github.com/users/geotrellis/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/geotrellis/subscriptions","organizations_url":"https://api.github.com/users/geotrellis/orgs","repos_url":"https://api.github.com/users/geotrellis/repos","events_url":"https://api.github.com/users/geotrellis/events{/privacy}","received_events_url":"https://api.github.com/users/geotrellis/received_events","type":"Organization","site_admin":false},"private":false,"html_url":"https://github.com/geotrellis/vectorpipe","description":"Convert Vector data to VectorTiles with GeoTrellis.","fork":false,"url":"https://api.github.com/repos/geotrellis/vectorpipe","forks_url":"https://api.github.com/repos/geotrellis/vectorpipe/forks","keys_url":"https://api.github.com/repos/geotrellis/vectorpipe/keys{/key_id}","collaborators_url":"https://api.github.com/repos/geotrellis/vectorpipe/collaborators{/collaborator}","teams_url":"https://api.github.com/repos/geotrellis/vectorpipe/teams","hooks_url":"https://api.github.com/repos/geotrellis/vectorpipe/hooks","issue_events_url":"https://api.github.com/repos/geotrellis/vectorpipe/issues/events{/number}","events_url":"https://api.github.com/repos/geotrellis/vectorpipe/events","assignees_url":"https://api.github.com/repos/geotrellis/vectorpipe/assignees{/user}","branches_url":"https://api.github.com/repos/geotrellis/vectorpipe/branches{/branch}","tags_url":"https://api.github.com/repos/geotrellis/vectorpipe/tags","blobs_url":"https://api.github.com/repos/geotrellis/vectorpipe/git/blobs{/sha}","git_tags_url":"https://api.github.com/repos/geotrellis/vectorpipe/git/tags{/sha}","git_refs_url":"https://api.github.com/repos/geotrellis/vectorpipe/git/refs{/sha}","trees_url":"https://api.github.com/repos/geotrellis/vectorpipe/git/trees{/sha}","statuses_url":"https://api.github.com/repos/geotrellis/vectorpipe/statuses/{sha}","languages_url":"https://api.github.com/repos/geotrellis/vectorpipe/languages","stargazers_url":"https://api.github.com/repos/geotrellis/vectorpipe/stargazers","contributors_url":"https://api.github.com/repos/geotrellis/vectorpipe/contributors","subscribers_url":"https://api.github.com/repos/geotrellis/vectorpipe/subscribers","subscription_url":"https://api.github.com/repos/geotrellis/vectorpipe/subscription","commits_url":"https://api.github.com/repos/geotrellis/vectorpipe/commits{/sha}","git_commits_url":"https://api.github.com/repos/geotrellis/vectorpipe/git/commits{/sha}","comments_url":"https://api.github.com/repos/geotrellis/vectorpipe/comments{/number}","issue_comment_url":"https://api.github.com/repos/geotrellis/vectorpipe/issues/comments{/number}","contents_url":"https://api.github.com/repos/geotrellis/vectorpipe/contents/{+path}","compare_url":"https://api.github.com/repos/geotrellis/vectorpipe/compare/{base}...{head}","merges_url":"https://api.github.com/repos/geotrellis/vectorpipe/merges","archive_url":"https://api.github.com/repos/geotrellis/vectorpipe/{archive_format}{/ref}","downloads_url":"https://api.github.com/repos/geotrellis/vectorpipe/downloads","issues_url":"https://api.github.com/repos/geotrellis/vectorpipe/issues{/number}","pulls_url":"https://api.github.com/repos/geotrellis/vectorpipe/pulls{/number}","milestones_url":"https://api.github.com/repos/geotrellis/vectorpipe/milestones{/number}","notifications_url":"https://api.github.com/repos/geotrellis/vectorpipe/notifications{?since,all,participating}","labels_url":"https://api.github.com/repos/geotrellis/vectorpipe/labels{/name}","releases_url":"https://api.github.com/repos/geotrellis/vectorpipe/releases{/id}","deployments_url":"https://api.github.com/repos/geotrellis/vectorpipe/deployments","created_at":"2016-10-20T18:35:16Z","updated_at":"2018-04-26T15:08:28Z","pushed_at":"2018-05-30T12:45:51Z","git_url":"git://github.com/geotrellis/vectorpipe.git","ssh_url":"git@github.com:geotrellis/vectorpipe.git","clone_url":"https://github.com/geotrellis/vectorpipe.git","svn_url":"https://github.com/geotrellis/vectorpipe","homepage":"https://geotrellis.github.io/vectorpipe/","size":10426,"stargazers_count":20,"watchers_count":20,"language":"Scala","has_issues":true,"has_projects":false,"has_downloads":true,"has_wiki":false,"has_pages":true,"forks_count":9,"mirror_url":null,"archived":false,"open_issues_count":8,"license":{"key":"other","name":"Other","spdx_id":null,"url":null},"forks":9,"open_issues":8,"watchers":20,"default_branch":"master"}},"_links":{"self":{"href":"https://api.github.com/repos/geotrellis/vectorpipe/pulls/54"},"html":{"href":"https://github.com/geotrellis/vectorpipe/pull/54"},"issue":{"href":"https://api.github.com/repos/geotrellis/vectorpipe/issues/54"},"comments":{"href":"https://api.github.com/repos/geotrellis/vectorpipe/issues/54/comments"},"review_comments":{"href":"https://api.github.com/repos/geotrellis/vectorpipe/pulls/54/comments"},"review_comment":{"href":"https://api.github.com/repos/geotrellis/vectorpipe/pulls/comments{/number}"},"commits":{"href":"https://api.github.com/repos/geotrellis/vectorpipe/pulls/54/commits"},"statuses":{"href":"https://api.github.com/repos/geotrellis/vectorpipe/statuses/c9f8e4a836ed8c8f5bc96eb78d9f1914fc7ae74f"}},"author_association":"NONE"}}
{"actor":{"display_login":"jbouffard"}}