Merge pull request #298 from s22s/fix/295

Removed typed column specification for ProjectedRasterTile functions.

Author: metasim

bench/src/main/scala/org/locationtech/rasterframes/bench/BinaryTileOpBench.scala

@@ -58,7 +58,7 @@ class BinaryTileOpBench extends SparkEnv {
 
   @Benchmark
   def viaExpression(): Array[Tile] = {
-    tiles.select(Add($"left", $"right")).collect()
+    tiles.select(Add($"left", $"right").as[Tile]).collect()
   }
 
   @Benchmark

bench/src/main/scala/org/locationtech/rasterframes/bench/TileEncodeBench.scala

@@ -55,7 +55,8 @@ class TileEncodeBench extends SparkEnv {
     cellTypeName match {
       case "rasterRef" ⇒
         val baseCOG = "https://s3-us-west-2.amazonaws.com/landsat-pds/c1/L8/149/039/LC08_L1TP_149039_20170411_20170415_01_T1/LC08_L1TP_149039_20170411_20170415_01_T1_B1.TIF"
-        tile = RasterRefTile(RasterRef(RasterSource(URI.create(baseCOG)), 0, Some(Extent(253785.0, 3235185.0, 485115.0, 3471015.0))))
+        val extent = Extent(253785.0, 3235185.0, 485115.0, 3471015.0)
+        tile = RasterRefTile(RasterRef(RasterSource(URI.create(baseCOG)), 0, Some(extent), None))
       case _ ⇒
         tile = randomTile(tileSize, tileSize, cellTypeName)
     }

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

@@ -73,8 +73,8 @@ trait RasterFunctions {
 
   @Experimental
   /** Convert array in `arrayCol` into a Tile of dimensions `cols` and `rows`*/
-  def rf_array_to_tile(arrayCol: Column, cols: Int, rows: Int) = withAlias("rf_array_to_tile", arrayCol)(
-    udf[Tile, AnyRef](F.arrayToTile(cols, rows)).apply(arrayCol)
+  def rf_array_to_tile(arrayCol: Column, cols: Int, rows: Int): TypedColumn[Any, Tile] = withTypedAlias("rf_array_to_tile")(
+    udf[Tile, AnyRef](F.arrayToTile(cols, rows)).apply(arrayCol).as[Tile]
   )
 
   /** Create a Tile from a column of cell data with location indexes and preform cell conversion. */
@@ -89,12 +89,10 @@ trait RasterFunctions {
   def rf_cell_type(col: Column): TypedColumn[Any, CellType] = GetCellType(col)
 
   /** Change the Tile's cell type */
-  def rf_convert_cell_type(col: Column, cellType: CellType): TypedColumn[Any, Tile] =
-    SetCellType(col, cellType)
+  def rf_convert_cell_type(col: Column, cellType: CellType): Column = SetCellType(col, cellType)
 
   /** Change the Tile's cell type */
-  def rf_convert_cell_type(col: Column, cellTypeName: String): TypedColumn[Any, Tile] =
-    SetCellType(col, cellTypeName)
+  def rf_convert_cell_type(col: Column, cellTypeName: String): Column = SetCellType(col, cellTypeName)
 
   /** 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. */
@@ -110,18 +108,20 @@ trait RasterFunctions {
   /** Extract the extent of a RasterSource or ProjectedRasterTile as a Geometry type. */
   def rf_geometry(raster: Column): TypedColumn[Any, Geometry] = GetGeometry(raster)
 
-  /** Assign a `NoData` value to the Tiles. */
-  def rf_with_no_data(col: Column, nodata: Double): TypedColumn[Any, Tile] = withTypedAlias("rf_with_no_data", col)(
-    udf[Tile, Tile](F.withNoData(nodata)).apply(col)
-  )
+  /** Assign a `NoData` value to the tile column. */
+  def rf_with_no_data(col: Column, nodata: Double): Column = SetNoDataValue(col, nodata)
+
+  /** Assign a `NoData` value to the tile column. */
+  def rf_with_no_data(col: Column, nodata: Int): Column = SetNoDataValue(col, nodata)
+
+  /** Assign a `NoData` value to the tile column. */
+  def rf_with_no_data(col: Column, nodata: Column): Column = SetNoDataValue(col, nodata)
 
   /**  Compute the full column aggregate floating point histogram. */
-  def rf_agg_approx_histogram(col: Column): TypedColumn[Any, CellHistogram] =
-    HistogramAggregate(col)
+  def rf_agg_approx_histogram(col: Column): TypedColumn[Any, CellHistogram] = HistogramAggregate(col)
 
   /** Compute the full column aggregate floating point statistics. */
-  def rf_agg_stats(col: Column): TypedColumn[Any, CellStatistics] =
-    CellStatsAggregate(col)
+  def rf_agg_stats(col: Column): TypedColumn[Any, CellStatistics] = CellStatsAggregate(col)
 
   /** Computes the column aggregate mean. */
   def rf_agg_mean(col: Column) = CellMeanAggregate(col)
@@ -194,28 +194,28 @@ trait RasterFunctions {
   def rf_agg_local_no_data_cells(col: Column): TypedColumn[Any, Tile] = LocalCountAggregate.LocalNoDataCellsUDAF(col)
 
   /** Cellwise addition between two Tiles or Tile and scalar column. */
-  def rf_local_add(left: Column, right: Column): TypedColumn[Any, Tile] = Add(left, right)
+  def rf_local_add(left: Column, right: Column): Column = Add(left, right)
 
   /** Cellwise addition of a scalar value to a tile. */
-  def rf_local_add[T: Numeric](tileCol: Column, value: T): TypedColumn[Any, Tile] = Add(tileCol, value)
+  def rf_local_add[T: Numeric](tileCol: Column, value: T): Column = Add(tileCol, value)
 
   /** Cellwise subtraction between two Tiles. */
-  def rf_local_subtract(left: Column, right: Column): TypedColumn[Any, Tile] = Subtract(left, right)
+  def rf_local_subtract(left: Column, right: Column): Column = Subtract(left, right)
 
   /** Cellwise subtraction of a scalar value from a tile. */
-  def rf_local_subtract[T: Numeric](tileCol: Column, value: T): TypedColumn[Any, Tile]  = Subtract(tileCol, value)
+  def rf_local_subtract[T: Numeric](tileCol: Column, value: T): Column  = Subtract(tileCol, value)
 
   /** Cellwise multiplication between two Tiles. */
-  def rf_local_multiply(left: Column, right: Column): TypedColumn[Any, Tile] = Multiply(left, right)
+  def rf_local_multiply(left: Column, right: Column): Column = Multiply(left, right)
 
   /** Cellwise multiplication of a tile by a scalar value. */
-  def rf_local_multiply[T: Numeric](tileCol: Column, value: T): TypedColumn[Any, Tile] = Multiply(tileCol, value)
+  def rf_local_multiply[T: Numeric](tileCol: Column, value: T): Column = Multiply(tileCol, value)
 
   /** Cellwise division between two Tiles. */
-  def rf_local_divide(left: Column, right: Column): TypedColumn[Any, Tile] = Divide(left, right)
+  def rf_local_divide(left: Column, right: Column): Column = Divide(left, right)
 
   /** Cellwise division of a tile by a scalar value. */
-  def rf_local_divide[T: Numeric](tileCol: Column, value: T): TypedColumn[Any, Tile] = Divide(tileCol, value)
+  def rf_local_divide[T: Numeric](tileCol: Column, value: T): Column = Divide(tileCol, value)
 
   /** Perform an arbitrary GeoTrellis `LocalTileBinaryOp` between two Tile columns. */
   def rf_local_algebra(op: LocalTileBinaryOp, left: Column, right: Column): TypedColumn[Any, Tile] =
@@ -231,9 +231,8 @@ trait RasterFunctions {
 
   /** Constructor for tile column with a single cell value. */
   def rf_make_constant_tile(value: Number, cols: Int, rows: Int, cellTypeName: String): TypedColumn[Any, Tile] = {
-    import org.apache.spark.sql.rf.TileUDT.tileSerializer
-    val constTile = encoders.serialized_literal(F.makeConstantTile(value, cols, rows, cellTypeName))
-    withTypedAlias(s"rf_make_constant_tile($value, $cols, $rows, $cellTypeName)")(constTile)
+    val constTile = udf(() => F.makeConstantTile(value, cols, rows, cellTypeName))
+    withTypedAlias(s"rf_make_constant_tile($value, $cols, $rows, $cellTypeName)")(constTile.apply())
   }
 
   /** Create a column constant tiles of zero */
@@ -326,96 +325,72 @@ trait RasterFunctions {
     DebugRender.RenderMatrix(col)
 
   /** Cellwise less than value comparison between two tiles. */
-  def rf_local_less(left: Column, right: Column): TypedColumn[Any, Tile] =
-    Less(left, right)
+  def rf_local_less(left: Column, right: Column): Column = Less(left, right)
 
   /** Cellwise less than value comparison between a tile and a scalar. */
-  def rf_local_less[T: Numeric](tileCol: Column, value: T): TypedColumn[Any, Tile] =
-    Less(tileCol, value)
+  def rf_local_less[T: Numeric](tileCol: Column, value: T): Column = Less(tileCol, value)
 
   /** Cellwise less than or equal to value comparison between a tile and a scalar. */
-  def rf_local_less_equal(left: Column, right: Column): TypedColumn[Any, Tile]  =
-    LessEqual(left, right)
+  def rf_local_less_equal(left: Column, right: Column): Column = LessEqual(left, right)
 
   /** Cellwise less than or equal to value comparison between a tile and a scalar. */
-  def rf_local_less_equal[T: Numeric](tileCol: Column, value: T): TypedColumn[Any, Tile] =
-    LessEqual(tileCol, value)
+  def rf_local_less_equal[T: Numeric](tileCol: Column, value: T): Column = LessEqual(tileCol, value)
 
   /** Cellwise greater than value comparison between two tiles. */
-  def rf_local_greater(left: Column, right: Column): TypedColumn[Any, Tile] =
-    Greater(left, right)
+  def rf_local_greater(left: Column, right: Column): Column = Greater(left, right)
 
   /** Cellwise greater than value comparison between a tile and a scalar. */
-  def rf_local_greater[T: Numeric](tileCol: Column, value: T): TypedColumn[Any, Tile] =
-    Greater(tileCol, value)
-
+  def rf_local_greater[T: Numeric](tileCol: Column, value: T): Column = Greater(tileCol, value)
   /** Cellwise greater than or equal to value comparison between two tiles. */
-  def rf_local_greater_equal(left: Column, right: Column): TypedColumn[Any, Tile]  =
-    GreaterEqual(left, right)
+  def rf_local_greater_equal(left: Column, right: Column): Column  = GreaterEqual(left, right)
 
   /** Cellwise greater than or equal to value comparison between a tile and a scalar. */
-  def rf_local_greater_equal[T: Numeric](tileCol: Column, value: T): TypedColumn[Any, Tile] =
-    GreaterEqual(tileCol, value)
+  def rf_local_greater_equal[T: Numeric](tileCol: Column, value: T): Column = GreaterEqual(tileCol, value)
 
   /** Cellwise equal to value comparison between two tiles. */
-  def rf_local_equal(left: Column, right: Column): TypedColumn[Any, Tile]  =
-    Equal(left, right)
+  def rf_local_equal(left: Column, right: Column): Column = Equal(left, right)
 
   /** Cellwise equal to value comparison between a tile and a scalar. */
-  def rf_local_equal[T: Numeric](tileCol: Column, value: T): TypedColumn[Any, Tile] =
-    Equal(tileCol, value)
+  def rf_local_equal[T: Numeric](tileCol: Column, value: T): Column = Equal(tileCol, value)
 
   /** Cellwise inequality comparison between two tiles. */
-  def rf_local_unequal(left: Column, right: Column): TypedColumn[Any, Tile]  =
-    Unequal(left, right)
+  def rf_local_unequal(left: Column, right: Column): Column  = Unequal(left, right)
 
   /** Cellwise inequality comparison between a tile and a scalar. */
-  def rf_local_unequal[T: Numeric](tileCol: Column, value: T): TypedColumn[Any, Tile] =
-    Unequal(tileCol, value)
+  def rf_local_unequal[T: Numeric](tileCol: Column, value: T): Column = Unequal(tileCol, value)
 
   /** Round cell values to nearest integer without chaning cell type. */
-  def rf_round(tileCol: Column): TypedColumn[Any, Tile] =
-    Round(tileCol)
+  def rf_round(tileCol: Column): Column = Round(tileCol)
 
   /** Compute the absolute value of each cell. */
-  def rf_abs(tileCol: Column): TypedColumn[Any, Tile] =
-    Abs(tileCol)
+  def rf_abs(tileCol: Column): Column = Abs(tileCol)
 
   /** Take natural logarithm of cell values. */
-  def rf_log(tileCol: Column): TypedColumn[Any, Tile] =
-    Log(tileCol)
+  def rf_log(tileCol: Column): Column = Log(tileCol)
 
   /** Take base 10 logarithm of cell values. */
-  def rf_log10(tileCol: Column): TypedColumn[Any, Tile] =
-    Log10(tileCol)
+  def rf_log10(tileCol: Column): Column = Log10(tileCol)
 
   /** Take base 2 logarithm of cell values. */
-  def rf_log2(tileCol: Column): TypedColumn[Any, Tile] =
-    Log2(tileCol)
+  def rf_log2(tileCol: Column): Column = Log2(tileCol)
 
   /** Natural logarithm of one plus cell values. */
-  def rf_log1p(tileCol: Column): TypedColumn[Any, Tile] =
-      Log1p(tileCol)
+  def rf_log1p(tileCol: Column): Column = Log1p(tileCol)
 
   /** Exponential of cell values */
-  def rf_exp(tileCol: Column): TypedColumn[Any, Tile] =
-    Exp(tileCol)
+  def rf_exp(tileCol: Column): Column = Exp(tileCol)
 
   /** Ten to the power of cell values */
-  def rf_exp10(tileCol: Column): TypedColumn[Any, Tile] =
-    Exp10(tileCol)
+  def rf_exp10(tileCol: Column): Column = Exp10(tileCol)
 
   /** Two to the power of cell values */
-  def rf_exp2(tileCol: Column): TypedColumn[Any, Tile] =
-    Exp2(tileCol)
+  def rf_exp2(tileCol: Column): Column = Exp2(tileCol)
 
   /** Exponential of cell values, less one*/
-  def rf_expm1(tileCol: Column): TypedColumn[Any, Tile] =
-    ExpM1(tileCol)
+  def rf_expm1(tileCol: Column): Column = ExpM1(tileCol)
 
   /** Return the incoming tile untouched. */
-  def rf_identity(tileCol: Column): TypedColumn[Any, Tile] =
-    Identity(tileCol)
+  def rf_identity(tileCol: Column): Column = Identity(tileCol)
 
   /** Create a row for each cell in Tile. */
   def rf_explode_tiles(cols: Column*): Column = rf_explode_tiles_sample(1.0, None, cols: _*)

core/src/main/scala/org/locationtech/rasterframes/encoders/StandardSerializers.scala

@@ -63,7 +63,30 @@ trait StandardSerializers {
       t.xmin, t.ymin, t.xmax, t.ymax
     )
     override def from[R](row: R, io: CatalystIO[R]): Extent = Extent(
-      io.getDouble(row, 0), io.getDouble(row, 1), io.getDouble(row, 2), io.getDouble(row, 3)
+      io.getDouble(row, 0),
+      io.getDouble(row, 1),
+      io.getDouble(row, 2),
+      io.getDouble(row, 3)
+    )
+  }
+
+  implicit val gridBoundsSerializer: CatalystSerializer[GridBounds] = new CatalystSerializer[GridBounds] {
+    override def schema: StructType = StructType(Seq(
+      StructField("colMin", IntegerType, false),
+      StructField("rowlMin", IntegerType, false),
+      StructField("colMax", IntegerType, false),
+      StructField("rowMax", IntegerType, false)
+    ))
+
+    override protected def to[R](t: GridBounds, io: CatalystIO[R]): R = io.create(
+      t.colMin, t.rowMin, t.colMax, t.rowMax
+    )
+
+    override protected def from[R](t: R, io: CatalystIO[R]): GridBounds = GridBounds(
+      io.getInt(t, 0),
+      io.getInt(t, 1),
+      io.getInt(t, 2),
+      io.getInt(t, 3)
     )
   }
 

core/src/main/scala/org/locationtech/rasterframes/expressions/generators/RasterSourceToRasterRefs.scala

@@ -22,6 +22,7 @@
 package org.locationtech.rasterframes.expressions.generators
 
 import com.typesafe.scalalogging.LazyLogging
+import geotrellis.raster.GridBounds
 import geotrellis.vector.Extent
 import org.apache.spark.sql.catalyst.InternalRow
 import org.apache.spark.sql.catalyst.expressions._
@@ -55,19 +56,22 @@ case class RasterSourceToRasterRefs(children: Seq[Expression], bandIndexes: Seq[
     name <- bandNames(basename, bandIndexes)
   } yield StructField(name, schemaOf[RasterRef], true))
 
-  private def band2ref(src: RasterSource, e: Option[Extent])(b: Int): RasterRef =
-    if (b < src.bandCount) RasterRef(src, b, e) else null
+  private def band2ref(src: RasterSource, e: Option[(GridBounds, Extent)])(b: Int): RasterRef =
+    if (b < src.bandCount) RasterRef(src, b, e.map(_._2), e.map(_._1)) else null
+
 
   override def eval(input: InternalRow): TraversableOnce[InternalRow] = {
     try {
       val refs = children.map { child ⇒
         val src = RasterSourceType.deserialize(child.eval(input))
-        subtileDims.map(dims =>
-          src
-            .layoutExtents(dims)
-            .map(e ⇒ bandIndexes.map(band2ref(src, Some(e))))
-        )
-        .getOrElse(Seq(bandIndexes.map(band2ref(src, None))))
+        val srcRE = src.rasterExtent
+        subtileDims.map(dims => {
+          val subGB = src.layoutBounds(dims)
+          val subs = subGB.map(gb => (gb, srcRE.extentFor(gb, clamp = true)))
+
+          subs.map(p => bandIndexes.map(band2ref(src, Some(p))))
+        })
+          .getOrElse(Seq(bandIndexes.map(band2ref(src, None))))
       }
       refs.transpose.map(ts ⇒ InternalRow(ts.flatMap(_.map(_.toInternalRow)): _*))
     }

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

@@ -22,10 +22,9 @@
 package org.locationtech.rasterframes.expressions.localops
 
 import geotrellis.raster.Tile
-import org.apache.spark.sql.catalyst.expressions.{Expression, ExpressionDescription}
+import org.apache.spark.sql.Column
 import org.apache.spark.sql.catalyst.expressions.codegen.CodegenFallback
-import org.apache.spark.sql.{Column, TypedColumn}
-import org.locationtech.rasterframes._
+import org.apache.spark.sql.catalyst.expressions.{Expression, ExpressionDescription}
 import org.locationtech.rasterframes.expressions.{NullToValue, UnaryLocalRasterOp}
 
 @ExpressionDescription(
@@ -45,6 +44,5 @@ case class Abs(child: Expression) extends UnaryLocalRasterOp with NullToValue wi
 }
 
 object Abs {
-  def apply(tile: Column): TypedColumn[Any, Tile] =
-    new Column(Abs(tile.expr)).as[Tile]
+  def apply(tile: Column): Column = new Column(Abs(tile.expr))
 }

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

@@ -22,12 +22,11 @@
 package org.locationtech.rasterframes.expressions.localops
 
 import geotrellis.raster.Tile
+import org.apache.spark.sql.Column
 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}
-import org.locationtech.rasterframes._
 import org.locationtech.rasterframes.expressions.BinaryLocalRasterOp
 import org.locationtech.rasterframes.expressions.DynamicExtractors.tileExtractor
 
@@ -65,9 +64,9 @@ case class Add(left: Expression, right: Expression) extends BinaryLocalRasterOp
   }
 }
 object Add {
-  def apply(left: Column, right: Column): TypedColumn[Any, Tile] =
-    new Column(Add(left.expr, right.expr)).as[Tile]
+  def apply(left: Column, right: Column): Column =
+    new Column(Add(left.expr, right.expr))
 
-  def apply[N: Numeric](tile: Column, value: N): TypedColumn[Any, Tile] =
-    new Column(Add(tile.expr, lit(value).expr)).as[Tile]
+  def apply[N: Numeric](tile: Column, value: N): Column =
+    new Column(Add(tile.expr, lit(value).expr))
 }

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

@@ -21,12 +21,11 @@
 
 package org.locationtech.rasterframes.expressions.localops
 import geotrellis.raster.Tile
+import org.apache.spark.sql.Column
 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}
-import org.locationtech.rasterframes._
 import org.locationtech.rasterframes.expressions.BinaryLocalRasterOp
 import org.locationtech.rasterframes.expressions.DynamicExtractors.tileExtractor
 import org.locationtech.rasterframes.util.DataBiasedOp
@@ -66,9 +65,9 @@ case class BiasedAdd(left: Expression, right: Expression) extends BinaryLocalRas
   }
 }
 object BiasedAdd {
-  def apply(left: Column, right: Column): TypedColumn[Any, Tile] =
-    new Column(BiasedAdd(left.expr, right.expr)).as[Tile]
+  def apply(left: Column, right: Column): Column =
+    new Column(BiasedAdd(left.expr, right.expr))
 
-  def apply[N: Numeric](tile: Column, value: N): TypedColumn[Any, Tile] =
-    new Column(BiasedAdd(tile.expr, lit(value).expr)).as[Tile]
+  def apply[N: Numeric](tile: Column, value: N): Column =
+    new Column(BiasedAdd(tile.expr, lit(value).expr))
 }