Pass layer name to clip()

CHANGELOG.md

@@ -24,6 +24,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 - Adds `riverbank`, `stream_end`, `dam`, `weir`, `waterfall`, and `pressurised`
   to the list of waterway features
 - Populates `nds` and `members` for deleted elements from the previous version
+- `Pipeline.clip()` now receives layer name
 
 ### Fixed
 

src/main/scala/vectorpipe/VectorPipe.scala

@@ -76,21 +76,21 @@ object VectorPipe {
 
     def generateVectorTiles[G <: Geometry](df: DataFrame, level: LayoutLevel): RDD[(SpatialKey, VectorTile)] = {
       val zoom = level.zoom
-      val clip = udf { (g: jts.Geometry, key: GenericRowWithSchema) =>
+      val clip = udf { (g: jts.Geometry, key: GenericRowWithSchema, layerName: String) =>
         val k = getSpatialKey(key)
-        pipeline.clip(g, k, level)
+        pipeline.clip(g, k, layerName, level)
       }
 
       val selectedGeometry = pipeline
         .select(df, zoom, keyColumn)
 
-      val clipped = selectedGeometry
-        .withColumn(keyColumn, explode(col(keyColumn)))
-        .repartition(col(keyColumn)) // spread copies of possibly ill-tempered geometries around cluster prior to clipping
-        .withColumn(geomColumn, clip(col(geomColumn), col(keyColumn)))
-
       pipeline.layerMultiplicity match {
         case SingleLayer(layerName) =>
+          val clipped = selectedGeometry
+            .withColumn(keyColumn, explode(col(keyColumn)))
+            .repartition(col(keyColumn)) // spread copies of possibly ill-tempered geometries around cluster prior to clipping
+            .withColumn(geomColumn, clip(col(geomColumn), col(keyColumn), lit(layerName)))
+
           clipped
             .rdd
             .map { r => (getSpatialKey(r, keyColumn), pipeline.pack(r, zoom)) }
@@ -102,6 +102,11 @@ object VectorPipe {
         case LayerNamesInColumn(layerNameCol) =>
           assert(selectedGeometry.schema(layerNameCol).dataType == StringType,
                  s"layerMultiplicity=${pipeline.layerMultiplicity} requires String-type column of name ${layerNameCol}")
+          val clipped = selectedGeometry
+            .withColumn(keyColumn, explode(col(keyColumn)))
+            .repartition(col(keyColumn)) // spread copies of possibly ill-tempered geometries around cluster prior to clipping
+            .withColumn(geomColumn, clip(col(geomColumn), col(keyColumn), col(layerNameCol)))
+
           clipped
             .rdd
             .map { r => (getSpatialKey(r, keyColumn), r.getAs[String](layerNameCol) -> pipeline.pack(r, zoom)) }

src/main/scala/vectorpipe/vectortile/Pipeline.scala

@@ -140,7 +140,7 @@ trait Pipeline {
    *
    * Basic (non-no-op) clipping functions can be found in [[Clipping]].
    */
-  def clip(geom: jts.Geometry, key: SpatialKey, layoutLevel: LayoutLevel): jts.Geometry = geom
+  def clip(geom: jts.Geometry, key: SpatialKey, layer: String, layoutLevel: LayoutLevel): jts.Geometry = geom
 
   /**
    * Convert table rows to output features.

src/test/scala/vectorpipe/vectortile/LayerTestPipeline.scala

@@ -20,6 +20,6 @@ case class LayerTestPipeline(geometryColumn: String, baseOutputURI: java.net.URI
       .where(functions.not(functions.isnull('layers)))
   }
 
-  override def clip(geom: jts.Geometry, key: geotrellis.spark.SpatialKey, layoutLevel: geotrellis.spark.tiling.LayoutLevel): jts.Geometry =
+  override def clip(geom: jts.Geometry, key: geotrellis.spark.SpatialKey, layer: String, layoutLevel: geotrellis.spark.tiling.LayoutLevel): jts.Geometry =
     Clipping.byLayoutCell(geom, key, layoutLevel)
 }