Add simple NN resample function

Signed-off-by: Jason T. Brown <jason@astraea.earth>

Author: vpipkt

core/src/main/scala/astraea/spark/rasterframes/RasterFunctions.scala

@@ -344,4 +344,10 @@ trait RasterFunctions {
   def expm1(tileCol: Column): TypedColumn[Any, Tile] =
     ExpM1(tileCol)
 
+  /** Resample tile using nearest-neighbor */
+  def resample[T: Numeric](tileCol: Column, value: T) = Resample(tileCol, value)
+
+  /** Resample tile using nearest-neighbor */
+  def resample(tileCol: Column, column2: Column) = Resample(tileCol, column2)
+
 }

core/src/main/scala/astraea/spark/rasterframes/expressions/localops/Resample.scala

@@ -0,0 +1,76 @@
+/*
+ * This software is licensed under the Apache 2 license, quoted below.
+ *
+ * Copyright 2019 Astraea, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License"); you may not
+ * use this file except in compliance with the License. You may obtain a copy of
+ * the License at
+ *
+ *     [http://www.apache.org/licenses/LICENSE-2.0]
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+ * License for the specific language governing permissions and limitations under
+ * the License.
+ *
+ * SPDX-License-Identifier: Apache-2.0
+ *
+ */
+
+package astraea.spark.rasterframes.expressions.localops
+
+import astraea.spark.rasterframes._
+import astraea.spark.rasterframes.expressions.DynamicExtractors.tileExtractor
+import astraea.spark.rasterframes.expressions.BinaryLocalRasterOp
+import geotrellis.raster.Tile
+import geotrellis.raster.resample.NearestNeighbor
+import org.apache.spark.sql.rf._
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.expressions.codegen.CodegenFallback
+import org.apache.spark.sql.catalyst.expressions.{Expression, ExpressionDescription}
+import org.apache.spark.sql.functions.lit
+import org.apache.spark.sql.{Column, TypedColumn}
+
+@ExpressionDescription(
+  usage = "_FUNC_(tile, factor) - Resample tile to different size based on scalar factor or tile whose dimension to match. Scalar less than one will downsample tile; greater than one will upsample. Uses nearest-neighbor value.",
+  arguments = """
+  Arguments:
+    * tile - tile
+    * rhs  - scalar or tile to match dimension""",
+  examples = """
+  Examples:
+    > SELECT _FUNC_(tile, 2.0);
+       ...
+    > SELECT _FUNC_(tile1, tile2);
+       ..."""
+)
+case class Resample(left: Expression, right: Expression) extends BinaryLocalRasterOp
+  with CodegenFallback {
+  override val nodeName: String = "resample"
+  override protected def op(left: Tile, right: Tile): Tile = left.resample(right.cols, right.rows, NearestNeighbor)
+  override protected def op(left: Tile, right: Double): Tile = left.resample((left.cols * right).toInt,
+                                                                             (left.rows * right).toInt, NearestNeighbor)
+  override protected def op(left: Tile, right: Int): Tile = op(left, right.toDouble)
+
+  override def eval(input: InternalRow): Any = {
+    if(input == null) null
+    else {
+      val l = left.eval(input)
+      val r = right.eval(input)
+      if (l == null && r == null) null
+      else if (l == null) r
+      else if (r == null && tileExtractor.isDefinedAt(right.dataType)) l
+      else if (r == null) null
+      else nullSafeEval(l, r)
+    }
+  }
+}
+object Resample{
+  def apply(left: Column, right: Column): TypedColumn[Any, Tile] =
+    new Column(Resample(left.expr, right.expr)).as[Tile]
+
+  def apply[N: Numeric](tile: Column, value: N): TypedColumn[Any, Tile] =
+    new Column(Resample(tile.expr, lit(value).expr)).as[Tile]
+}

core/src/main/scala/astraea/spark/rasterframes/expressions/package.scala

@@ -86,6 +86,7 @@ package object expressions {
     registry.registerExpression[Exp10]("rf_exp10")
     registry.registerExpression[Exp2]("rf_exp2")
     registry.registerExpression[ExpM1]("rf_expm1")
+    registry.registerExpression[Resample]("rf_resample")
     registry.registerExpression[TileToArrayDouble]("rf_tile_to_array_double")
     registry.registerExpression[TileToArrayInt]("rf_tile_to_array_int")
     registry.registerExpression[DataCells]("rf_data_cells")

core/src/test/scala/astraea/spark/rasterframes/RasterFunctionsSpec.scala

@@ -27,7 +27,7 @@ import astraea.spark.rasterframes.tiles.ProjectedRasterTile
 import geotrellis.proj4.LatLng
 import geotrellis.raster
 import geotrellis.raster.testkit.RasterMatchers
-import geotrellis.raster.{BitCellType, ByteUserDefinedNoDataCellType, DoubleConstantNoDataCellType, ShortConstantNoDataCellType, Tile, UByteConstantNoDataCellType}
+import geotrellis.raster.{ArrayTile, BitCellType, ByteUserDefinedNoDataCellType, DoubleConstantNoDataCellType, ShortConstantNoDataCellType, Tile, UByteConstantNoDataCellType}
 import geotrellis.vector.Extent
 import org.apache.spark.sql.{AnalysisException, Encoders}
 import org.apache.spark.sql.functions._
@@ -659,4 +659,37 @@ class RasterFunctionsSpec extends FunSpec
 
     }
   }
+  it("should resample") {
+    def lowRes = {
+      def base = ArrayTile(Array(1,2,3,4), 2, 2)
+      ProjectedRasterTile(base.convert(ct), extent, crs)
+    }
+    def upsampled = {
+      def base = ArrayTile(Array(
+        1,1,2,2,
+        1,1,2,2,
+        3,3,4,4,
+        3,3,4,4
+      ), 4, 4)
+      ProjectedRasterTile(base.convert(ct), extent, crs)
+    }
+    // a 4, 4 tile to upsample by shape
+    def fourByFour = TestData.projectedRasterTile(4, 4, 0, extent, crs, ct)
+
+    def df = Seq(lowRes).toDF("tile")
+
+    val maybeUp = df.select(resample($"tile", lit(2))).as[ProjectedRasterTile].first()
+    assertEqual(maybeUp, upsampled)
+
+    def df2 = Seq((lowRes, fourByFour)).toDF("tile1", "tile2")
+    val maybeUpShape = df2.select(resample($"tile1", $"tile2")).as[ProjectedRasterTile].first()
+    assertEqual(maybeUpShape, upsampled)
+
+    // Downsample by double argument < 1
+    def df3 = Seq(upsampled).toDF("tile").withColumn("factor", lit(0.5))
+    assertEqual(df3.selectExpr("rf_resample(tile, 0.5)").as[ProjectedRasterTile].first(), lowRes)
+    assertEqual(df3.selectExpr("rf_resample(tile, factor)").as[ProjectedRasterTile].first(), lowRes)
+
+    checkDocs("rf_resample")
+  }
 }