PR feedback

Signed-off-by: Jason T. Brown <[email protected]>

Author: vpipkt

core/src/main/scala/org/locationtech/rasterframes/expressions/localops/Resample.scala

@@ -23,7 +23,7 @@ package org.locationtech.rasterframes.expressions.localops
 
 import geotrellis.raster.Tile
 import geotrellis.raster.resample._
-import geotrellis.raster.resample.{Max ⇒ RMax, Min ⇒ RMin}
+import geotrellis.raster.resample.{ResampleMethod ⇒ GTResampleMethod, Max ⇒ RMax, Min ⇒ RMin}
 import org.apache.spark.sql.Column
 import org.apache.spark.sql.catalyst.InternalRow
 import org.apache.spark.sql.catalyst.analysis.TypeCheckResult
@@ -34,6 +34,7 @@ import org.apache.spark.sql.functions.lit
 import org.apache.spark.sql.rf.TileUDT
 import org.apache.spark.sql.types.{DataType, StringType}
 import org.apache.spark.unsafe.types.UTF8String
+import org.locationtech.rasterframes.util.ResampleMethod
 import org.locationtech.rasterframes.encoders.CatalystSerializer._
 import org.locationtech.rasterframes.expressions.{fpTile, row}
 import org.locationtech.rasterframes.expressions.DynamicExtractors._
@@ -47,40 +48,21 @@ abstract class ResampleBase(left: Expression, right: Expression, method: Express
   override def dataType: DataType = left.dataType
   override def children: Seq[Expression] = Seq(left, right, method)
 
-  def targetFloatIfNeeded(t: Tile, method: ResampleMethod): Tile =
+  def targetFloatIfNeeded(t: Tile, method: GTResampleMethod): Tile =
      method match {
       case NearestNeighbor | Mode | RMax | RMin | Sum  ⇒ t
       case _ ⇒ fpTile(t)
      }
 
-  def stringToMethod(methodName: String): ResampleMethod =
-    methodName.toLowerCase().trim().replaceAll("_", "") match {
-      case "nearestneighbor" | "nearest" ⇒ NearestNeighbor
-      case "bilinear" ⇒ Bilinear
-      case "cubicconvolution" ⇒ CubicConvolution
-      case "cubicspline" ⇒ CubicSpline
-      case "lanczos" | "lanzos" ⇒ Lanczos
-      // aggregates
-      case "average" ⇒ Average
-      case "mode" ⇒ Mode
-      case "median" ⇒ Median
-      case "max" ⇒ RMax
-      case "min" ⇒ RMin
-      case "sum" ⇒ Sum
-    }
-
    // These methods define the core algorithms to be used.
-  def op(left: Tile, right: Tile, method: String): Tile = {
-    val m = stringToMethod(method)
-    targetFloatIfNeeded(left, m)
-      .resample(right.cols, right.rows, m)
-  }
+  def op(left: Tile, right: Tile, method: GTResampleMethod): Tile =
+    op(left, right.cols, right.rows, method)
 
-  def op(left: Tile, right: Double, method: String): Tile = {
-    val m = stringToMethod(method)
-    targetFloatIfNeeded(left, m)
-      .resample((left.cols * right).toInt, (left.rows * right).toInt, m)
-  }
+  def op(left: Tile, right: Double, method: GTResampleMethod): Tile =
+    op(left, (left.cols * right).toInt, (left.rows * right).toInt, method)
+
+  def op(tile: Tile, newCols: Int, newRows: Int, method: GTResampleMethod): Tile =
+    targetFloatIfNeeded(tile, method).resample(newCols, newRows, method)
 
   override def checkInputDataTypes(): TypeCheckResult = {
     // copypasta from BinaryLocalRasterOp
@@ -102,11 +84,16 @@ abstract class ResampleBase(left: Expression, right: Expression, method: Express
     val (leftTile, leftCtx) = tileExtractor(left.dataType)(row(input1))
     val methodString = input3.asInstanceOf[UTF8String].toString
 
+    val resamplingMethod = methodString match {
+      case ResampleMethod(mm) => mm
+      case _ => throw new IllegalArgumentException("Unrecognized resampling method specified")
+    }
+
     val result: Tile = tileOrNumberExtractor(right.dataType)(input2) match {
       // in this case we expect the left and right contexts to vary. no warnings raised.
-      case TileArg(rightTile, _) ⇒ op(leftTile, rightTile, methodString)
-      case DoubleArg(d) ⇒ op(leftTile, d, methodString)
-      case IntegerArg(i) ⇒ op(leftTile, i.toDouble, methodString)
+      case TileArg(rightTile, _) ⇒ op(leftTile, rightTile, resamplingMethod)
+      case DoubleArg(d) ⇒ op(leftTile, d, resamplingMethod)
+      case IntegerArg(i) ⇒ op(leftTile, i.toDouble, resamplingMethod)
     }
 
     // reassemble the leftTile with its context. Note that this operation does not change Extent and CRS
@@ -177,7 +164,9 @@ object Resample {
     > SELECT _FUNC_(tile1, tile2);
        ...""")
 case class ResampleNearest(tile: Expression, target: Expression)
-  extends ResampleBase(tile, target, Literal("nearest"))
+  extends ResampleBase(tile, target, Literal("nearest")) {
+  override val nodeName: String = "rf_resample_nearest"
+}
 object ResampleNearest {
   def apply(tile: Column, target: Column): Column =
     new Column(ResampleNearest(tile.expr, target.expr))

core/src/main/scala/org/locationtech/rasterframes/extensions/DataFrameMethods.scala

@@ -23,6 +23,7 @@ package org.locationtech.rasterframes.extensions
 
 import geotrellis.proj4.CRS
 import geotrellis.layer._
+import geotrellis.raster.resample.{NearestNeighbor, ResampleMethod => GTResampleMethod}
 import geotrellis.util.MethodExtensions
 import geotrellis.vector.Extent
 import org.apache.spark.sql.catalyst.expressions.Attribute
@@ -37,6 +38,7 @@ import org.locationtech.rasterframes.util._
 import org.locationtech.rasterframes.{MetadataKeys, RasterFrameLayer}
 import spray.json.JsonFormat
 import org.locationtech.rasterframes.util.JsonCodecs._
+
 import scala.util.Try
 
 /**
@@ -168,7 +170,7 @@ trait DataFrameMethods[DF <: DataFrame] extends MethodExtensions[DF] with Metada
     * @param resampleMethod string indicating method to use for resampling.
     * @return joined dataframe
     */
-  def rasterJoin(right: DataFrame, resampleMethod: String = "nearest"): DataFrame = RasterJoin(self, right, resampleMethod, None)
+  def rasterJoin(right: DataFrame, resampleMethod: GTResampleMethod = NearestNeighbor): DataFrame = RasterJoin(self, right, resampleMethod, None)
 
   /**
     * Performs a jeft join on the dataframe `right` to this one, reprojecting and merging tiles as necessary.
@@ -187,7 +189,7 @@ trait DataFrameMethods[DF <: DataFrame] extends MethodExtensions[DF] with Metada
     * @param resampleMethod string indicating method to use for resampling.
     * @return joined dataframe
     */
-  def rasterJoin(right: DataFrame, leftExtent: Column, leftCRS: Column, rightExtent: Column, rightCRS: Column, resampleMethod: String): DataFrame =
+  def rasterJoin(right: DataFrame, leftExtent: Column, leftCRS: Column, rightExtent: Column, rightCRS: Column, resampleMethod: GTResampleMethod): DataFrame =
     RasterJoin(self, right, leftExtent, leftCRS, rightExtent, rightCRS, resampleMethod, None)
 
   /**
@@ -205,7 +207,7 @@ trait DataFrameMethods[DF <: DataFrame] extends MethodExtensions[DF] with Metada
     * @param resampleMethod string indicating method to use for resampling.
     * @return joined dataframe
     */
-  def rasterJoin(right: DataFrame, joinExpr: Column, leftExtent: Column, leftCRS: Column, rightExtent: Column, rightCRS: Column, resampleMethod: String): DataFrame =
+  def rasterJoin(right: DataFrame, joinExpr: Column, leftExtent: Column, leftCRS: Column, rightExtent: Column, rightCRS: Column, resampleMethod: GTResampleMethod): DataFrame =
     RasterJoin(self, right, joinExpr, leftExtent, leftCRS, rightExtent, rightCRS, resampleMethod, None)
 
 

core/src/main/scala/org/locationtech/rasterframes/extensions/RasterJoin.scala

@@ -21,6 +21,7 @@
 
 package org.locationtech.rasterframes.extensions
 import geotrellis.raster.Dimensions
+import geotrellis.raster.resample.{NearestNeighbor, ResampleMethod => GTResampleMethod}
 import org.apache.spark.sql._
 import org.apache.spark.sql.functions._
 import org.apache.spark.sql.types.DataType
@@ -36,7 +37,7 @@ import scala.util.Random
 object RasterJoin {
 
   /** Perform a raster join on dataframes that each have proj_raster columns, or crs and extent explicitly included. */
-  def apply(left: DataFrame, right: DataFrame, resampleMethod: String, fallbackDimensions: Option[Dimensions[Int]]): DataFrame = {
+  def apply(left: DataFrame, right: DataFrame, resampleMethod: GTResampleMethod, fallbackDimensions: Option[Dimensions[Int]]): DataFrame = {
     def usePRT(d: DataFrame) =
       d.projRasterColumns.headOption
         .map(p => (rf_crs(p),  rf_extent(p)))
@@ -53,7 +54,7 @@ object RasterJoin {
     apply(ldf, rdf, lextent, lcrs, rextent, rcrs, resampleMethod, fallbackDimensions)
   }
 
-  def apply(left: DataFrame, right: DataFrame, leftExtent: Column, leftCRS: Column, rightExtent: Column, rightCRS: Column, resampleMethod: String, fallbackDimensions: Option[Dimensions[Int]]): DataFrame = {
+  def apply(left: DataFrame, right: DataFrame, leftExtent: Column, leftCRS: Column, rightExtent: Column, rightCRS: Column, resampleMethod: GTResampleMethod, fallbackDimensions: Option[Dimensions[Int]]): DataFrame = {
     val leftGeom = st_geometry(leftExtent)
     val rightGeomReproj = st_reproject(st_geometry(rightExtent), rightCRS, leftCRS)
     val joinExpr = new Column(SpatialRelation.Intersects(leftGeom.expr, rightGeomReproj.expr))
@@ -64,7 +65,7 @@ object RasterJoin {
     require(extractor.isDefinedAt(col.expr.dataType), s"Expected column ${col} to be of type $description, but was ${col.expr.dataType}.")
   }
 
-  def apply(left: DataFrame, right: DataFrame, joinExprs: Column, leftExtent: Column, leftCRS: Column, rightExtent: Column, rightCRS: Column, resampleMethod: String = "nearest", fallbackDimensions: Option[Dimensions[Int]] = None): DataFrame = {
+  def apply(left: DataFrame, right: DataFrame, joinExprs: Column, leftExtent: Column, leftCRS: Column, rightExtent: Column, rightCRS: Column, resampleMethod: GTResampleMethod = NearestNeighbor, fallbackDimensions: Option[Dimensions[Int]] = None): DataFrame = {
     // Convert resolved column into a symbolic one.
     def unresolved(c: Column): Column = col(c.columnName)
 
@@ -84,14 +85,13 @@ object RasterJoin {
     val rightExtent2 = id + "extent"
     // Post aggregation right crs. We create a new name.
     val rightCRS2 = id + "crs"
-    val method = id + "method"
 
     // Gathering up various expressions we'll use to construct the result.
     // After joining We will be doing a groupBy the LHS. We have to define the aggregations to perform after the groupBy.
     // On the LHS we just want the first thing (subsequent ones should be identical.
     val leftAggCols = left.columns.map(s => first(left(s), true) as s)
     // On the RHS we collect result as a list.
-    val rightAggCtx = Seq(collect_list(rightExtent) as rightExtent2, collect_list(rf_crs(rightCRS)) as rightCRS2, lit(resampleMethod) as method)
+    val rightAggCtx = Seq(collect_list(rightExtent) as rightExtent2, collect_list(rf_crs(rightCRS)) as rightCRS2)
     val rightAggTiles = right.tileColumns.map(c => collect_list(ExtractTile(c)) as c.columnName)
     val rightAggOther = right.notTileColumns
       .filter(n => n.columnName != rightExtent.columnName && n.columnName != rightCRS.columnName)
@@ -110,7 +110,7 @@ object RasterJoin {
 
     val reprojCols = rightAggTiles.map(t => {
       reproject_and_merge(
-        col(leftExtent2), col(leftCRS2), col(t.columnName), col(rightExtent2), col(rightCRS2), destDims, lit(resampleMethod)
+        col(leftExtent2), col(leftCRS2), col(t.columnName), col(rightExtent2), col(rightCRS2), destDims, lit(ResampleMethod(resampleMethod))
       ) as t.columnName
     })
 

core/src/main/scala/org/locationtech/rasterframes/extensions/ReprojectToLayer.scala

@@ -22,6 +22,7 @@
 package org.locationtech.rasterframes.extensions
 
 import geotrellis.layer._
+import geotrellis.raster.resample.{NearestNeighbor, ResampleMethod => GTResampleMethod}
 import org.apache.spark.sql._
 import org.apache.spark.sql.functions.broadcast
 import org.locationtech.rasterframes._
@@ -30,7 +31,7 @@ import org.locationtech.rasterframes.util._
 
 /** Algorithm for projecting an arbitrary RasterFrame into a layer with consistent CRS and gridding. */
 object ReprojectToLayer {
-  def apply(df: DataFrame, tlm: TileLayerMetadata[SpatialKey]): RasterFrameLayer = {
+  def apply(df: DataFrame, tlm: TileLayerMetadata[SpatialKey], resampleMethod: Option[GTResampleMethod] = None): RasterFrameLayer = {
     // create a destination dataframe with crs and extend columns
     // use RasterJoin to do the rest.
     val gb = tlm.tileBounds
@@ -48,7 +49,7 @@ object ReprojectToLayer {
     // Create effectively a target RasterFrame, but with no tiles.
     val dest = gridItems.toSeq.toDF(SPATIAL_KEY_COLUMN.columnName, EXTENT_COLUMN.columnName, CRS_COLUMN.columnName)
 
-    val joined = RasterJoin(broadcast(dest), df, "nearest", Some(tlm.tileLayout.tileDimensions))
+    val joined = RasterJoin(broadcast(dest), df, resampleMethod.getOrElse(NearestNeighbor), Some(tlm.tileLayout.tileDimensions))
 
     joined.asLayer(SPATIAL_KEY_COLUMN, tlm)
   }

core/src/main/scala/org/locationtech/rasterframes/package.scala

@@ -23,6 +23,7 @@ package org.locationtech
 import com.typesafe.config.ConfigFactory
 import com.typesafe.scalalogging.Logger
 import geotrellis.raster.{Dimensions, Tile, TileFeature, isData}
+import geotrellis.raster.resample._
 import geotrellis.layer._
 import geotrellis.spark.ContextRDD
 import org.apache.spark.rdd.RDD
@@ -147,4 +148,7 @@ package object rasterframes extends StandardColumns
     else isCellTrue(t.get(col, row))
 
 
+
+
+
 }

core/src/main/scala/org/locationtech/rasterframes/util/package.scala

@@ -186,6 +186,43 @@ package object util extends DataFrameRenderers {
     def apply() = mapping.keys.toSeq
   }
 
+  object ResampleMethod {
+    import geotrellis.raster.resample.{ResampleMethod ⇒ GTResampleMethod, _}
+    def unapply(name: String): Option[GTResampleMethod] = {
+      name.toLowerCase().trim().replaceAll("_", "") match {
+        case "nearestneighbor" | "nearest" ⇒ Some(NearestNeighbor)
+        case "bilinear" ⇒ Some(Bilinear)
+        case "cubicconvolution" ⇒ Some(CubicConvolution)
+        case "cubicspline" ⇒ Some(CubicSpline)
+        case "lanczos" | "lanzos" ⇒ Some(Lanczos)
+        // aggregates
+        case "average" ⇒ Some(Average)
+        case "mode" ⇒ Some(Mode)
+        case "median" ⇒ Some(Median)
+        case "max" ⇒ Some(Max)
+        case "min" ⇒ Some(Min)
+        case "sum" ⇒ Some(Sum)
+        case _ => None
+      }
+    }
+    def apply(gtr: GTResampleMethod): String = {
+      gtr match {
+        case NearestNeighbor ⇒ "nearest"
+        case Bilinear ⇒ "bilinear"
+        case CubicConvolution ⇒ "cubicconvolution"
+        case CubicSpline ⇒ "cubicspline"
+        case Lanczos ⇒ "lanczos"
+        case Average ⇒ "average"
+        case Mode ⇒ "mode"
+        case Median ⇒ "median"
+        case Max ⇒ "max"
+        case Min ⇒ "min"
+        case Sum ⇒ "sum"
+        case _ ⇒ throw new IllegalArgumentException(s"Unrecogized ResampleMethod ${gtr.toString()}")
+      }
+    }
+  }
+
   private[rasterframes]
   def toParquetFriendlyColumnName(name: String) = name.replaceAll("[ ,;{}()\n\t=]", "_")
 

core/src/test/scala/org/locationtech/rasterframes/RasterJoinSpec.scala

@@ -21,7 +21,7 @@
 
 package org.locationtech.rasterframes
 
-import geotrellis.raster.resample.Bilinear
+import geotrellis.raster.resample._
 import geotrellis.raster.testkit.RasterMatchers
 import geotrellis.raster.{Dimensions, IntConstantNoDataCellType, Raster, Tile}
 import org.apache.spark.SparkConf
@@ -175,8 +175,8 @@ class RasterJoinSpec extends TestEnvironment with TestData with RasterMatchers {
     it("should honor resampling options") {
       // test case. replicate existing test condition and check that resampling option results in different output
       val filterExpr = st_intersects(rf_geometry($"tile"), st_point(704940.0, 4251130.0))
-      val result = b4nativeRf.rasterJoin(b4warpedRf.withColumnRenamed("tile2", "nearest"), "nearest")
-        .rasterJoin(b4warpedRf.withColumnRenamed("tile2", "CubicSpline"), "cubicSpline")
+      val result = b4nativeRf.rasterJoin(b4warpedRf.withColumnRenamed("tile2", "nearest"), NearestNeighbor)
+        .rasterJoin(b4warpedRf.withColumnRenamed("tile2", "CubicSpline"), CubicSpline)
         .withColumn("diff", rf_local_subtract($"nearest", $"cubicSpline"))
         .agg(rf_agg_stats($"diff") as "stats")
         .select($"stats.min" as "min", $"stats.max" as "max")

docs/src/main/paradox/reference.md

@@ -168,9 +168,14 @@ In __SQL__, three parameters are required for `rf_resample`.:
     Tile rf_resample_nearest(Tile tile, Tile shape_tile)
 
 
-Change the tile dimension by upsampling or downsampling. Passing a numeric `factor` will scale the number of columns and rows in the tile: 1.0 is the same number of columns and row; less than one downsamples the tile; and greater than one upsamples the tile. Passing a tile as the second argument resamples such that the output has the same dimension (number of columns and rows) as `shape_tile`. Resampling methods can be one of: nearest_neighbor, bilinear, cubic_convolution, cubic_spline, lanczos, average, mode, median, max, min, or sum.
+Change the tile dimension by upsampling or downsampling. Passing a numeric `factor` will scale the number of columns and rows in the tile: 1.0 is the same number of columns and row; less than one downsamples the tile; and greater than one upsamples the tile. Passing a tile as the second argument resamples such that the output has the same dimension (number of columns and rows) as `shape_tile`. 
 
-Note the last six options apply aggregates when downsampling. For example a 0.25 factor and `max` method returns the maximum value in a 4x4 neighborhood.
+There are two categories: point resampling methods and aggregating resampling methods. 
+Resampling method to use can be specified by one of the following strings, possibly in a column.
+The point resampling methods are: `"nearest_neighbor"`, `"bilinear"`, `"cubic_convolution"`, `"cubic_spline"`, and `"lanczos"`.
+The aggregating resampling methods are: `"average"`, `"mode"`, `"median"`, `"max"`, "`min`", or `"sum"`.
+
+Note the aggregating methods are intended for downsampling. For example a 0.25 factor and `max` method returns the maximum value in a 4x4 neighborhood.
 
 If `tile` has an integer `CellType`, the returned tile will be coerced to a floating point with the following methods: bilinear, cubic_convolution, cubic_spline, lanczos, average, and median.
 

docs/src/main/paradox/release-notes.md

@@ -6,7 +6,7 @@
 
 * Added `method_name` parameter to the `rf_resample` method.
    * __BREAKING__: In SQL, the function `rf_resample` now takes 3 arguments. You can use `rf_resample_nearest` with two arguments or refactor to `rf_resample(t, v, "nearest")`.
-* Added resample method parameter to SQL and Python APIs. This will affect the reprojection of right hand side tiles.
+* Added resample method parameter to SQL and Python APIs. @ref:[See updated docs](raster-join.md).
 
 
 ### 0.9.0

pyrasterframes/src/main/python/docs/raster-join.pymd

@@ -53,7 +53,8 @@ The following optional arguments are allowed:
  * `right_extent` - the column on the right-hand DataFrame giving the [extent][extent] of the tile columns
  * `right_crs` - the column on the right-hand DataFrame giving the [CRS][CRS] of the tile columns
  * `join_exprs` - a single column expression as would be used in the [`on` parameter of `join`](https://spark.apache.org/docs/latest/api/python/pyspark.sql.html#pyspark.sql.DataFrame.join)
- * `resampling_method` - resampling algorithm to use in reprojection of right-hand tile columns. A string that is one of:nearest_neighbor, bilinear, cubic_convolution, cubic_spline, lanczos, average, mode, median, max, min, or sum.
+ * `resampling_method` - resampling algorithm to use in reprojection of right-hand tile column
+ 
 
 
  Note that the `join_exprs` will override the join behavior described above. By default the expression is equivalent to:
@@ -65,6 +66,10 @@ st_intersects(
 )
 ```
 
+Resampling method to use can be specified by passing one of the following strings into `resampling_method` parameter.
+The point resampling methods are: `"nearest_neighbor"`, `"bilinear"`, `"cubic_convolution"`, `"cubic_spline"`, and `"lanczos"`.
+The aggregating resampling methods are: `"average"`, `"mode"`, `"median"`, `"max"`, "`min`", or `"sum"`.
+Note the aggregating methods are intended for downsampling. For example a 0.25 factor and `max` method returns the maximum value in a 4x4 neighborhood.
 
 
 [CRS]: concepts.md#coordinate-reference-system--crs