Additional fixes for issue #242 that cropped back up.

Author: metasim

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/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,18 +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))
+        val srcRE = src.rasterExtent
         subtileDims.map(dims => {
-          val extents = src.layoutExtents(dims)
-          extents.map(e ⇒ bandIndexes.map(band2ref(src, Some(e))))
+          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))))
+          .getOrElse(Seq(bandIndexes.map(band2ref(src, None))))
       }
       refs.transpose.map(ts ⇒ InternalRow(ts.flatMap(_.map(_.toInternalRow)): _*))
     }

core/src/main/scala/org/locationtech/rasterframes/expressions/transformers/SetCellType.scala

@@ -50,7 +50,7 @@ import org.locationtech.rasterframes.expressions.row
     * rhs  - a cell type definition""",
   examples = """
   Examples:
-    > SELECT _FUNC_(tile, 1.5);
+    > SELECT _FUNC_(tile, 'int16ud0');
        ..."""
 )
 case class SetCellType(tile: Expression, cellType: Expression)

core/src/main/scala/org/locationtech/rasterframes/ref/RasterRef.scala

@@ -38,7 +38,7 @@ import org.locationtech.rasterframes.tiles.ProjectedRasterTile
  *
  * @since 8/21/18
  */
-case class RasterRef(source: RasterSource, bandIndex: Int, subextent: Option[Extent])
+case class RasterRef(source: RasterSource, bandIndex: Int, subextent: Option[Extent], subgrid: Option[GridBounds])
   extends ProjectedRasterLike {
   def crs: CRS = source.crs
   def extent: Extent = subextent.getOrElse(source.extent)
@@ -48,12 +48,12 @@ case class RasterRef(source: RasterSource, bandIndex: Int, subextent: Option[Ext
   def cellType: CellType = source.cellType
   def tile: ProjectedRasterTile = ProjectedRasterTile(RasterRefTile(this), extent, crs)
 
-  protected lazy val grid: GridBounds = source.rasterExtent.gridBoundsFor(extent, true)
-  protected def srcExtent: Extent = extent
+  protected lazy val grid: GridBounds =
+    subgrid.getOrElse(source.rasterExtent.gridBoundsFor(extent, true))
 
   protected lazy val realizedTile: Tile = {
-    RasterRef.log.trace(s"Fetching $srcExtent from band $bandIndex of $source")
-    source.read(srcExtent, Seq(bandIndex)).tile.band(0)
+    RasterRef.log.trace(s"Fetching $extent ($grid) from band $bandIndex of $source")
+    source.read(grid, Seq(bandIndex)).tile.band(0)
   }
 }
 
@@ -79,20 +79,24 @@ object RasterRef extends LazyLogging {
     override def schema: StructType = StructType(Seq(
       StructField("source", rsType, false),
       StructField("bandIndex", IntegerType, false),
-      StructField("subextent", schemaOf[Extent], true)
+      StructField("subextent", schemaOf[Extent], true),
+      StructField("subgrid", schemaOf[GridBounds], true)
     ))
 
     override def to[R](t: RasterRef, io: CatalystIO[R]): R = io.create(
       io.to(t.source)(RasterSourceUDT.rasterSourceSerializer),
       t.bandIndex,
-      t.subextent.map(io.to[Extent]).orNull
+      t.subextent.map(io.to[Extent]).orNull,
+      t.subgrid.map(io.to[GridBounds]).orNull
     )
 
     override def from[R](row: R, io: CatalystIO[R]): RasterRef = RasterRef(
       io.get[RasterSource](row, 0)(RasterSourceUDT.rasterSourceSerializer),
       io.getInt(row, 1),
       if (io.isNullAt(row, 2)) None
-      else Option(io.get[Extent](row, 2))
+      else Option(io.get[Extent](row, 2)),
+      if (io.isNullAt(row, 3)) None
+      else Option(io.get[GridBounds](row, 3))
     )
   }
 

core/src/main/scala/org/locationtech/rasterframes/ref/RasterSource.scala

@@ -78,7 +78,7 @@ trait RasterSource extends ProjectedRasterLike with Serializable {
 
   def layoutExtents(dims: TileDimensions): Seq[Extent] = {
     val re = rasterExtent
-    layoutBounds(dims).map(re.rasterExtentFor).map(_.extent)
+    layoutBounds(dims).map(re.extentFor(_, clamp = true))
   }
 
   def layoutBounds(dims: TileDimensions): Seq[GridBounds] = {

core/src/main/scala/org/locationtech/rasterframes/tiles/ProjectedRasterTile.scala

@@ -60,6 +60,7 @@ object ProjectedRasterTile {
       extends ProjectedRasterTile {
     def delegate: Tile = t
 
+    // NB: Don't be tempted to move this into the parent trait. Will get stack overflow.
     override def convert(cellType: CellType): Tile =
       ConcreteProjectedRasterTile(t.convert(cellType), extent, crs)
 

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

@@ -64,7 +64,7 @@ class RasterFunctionsSpec extends TestEnvironment with RasterMatchers {
     TestData.randomTile(cols, rows, UByteConstantNoDataCellType)
   )).map(ProjectedRasterTile(_, extent, crs)) :+ null
 
-  def lazyPRT = RasterRef(RasterSource(TestData.l8samplePath), 0, None).tile
+  def lazyPRT = RasterRef(RasterSource(TestData.l8samplePath), 0, None, None).tile
 
   implicit val pairEnc = Encoders.tuple(ProjectedRasterTile.prtEncoder, ProjectedRasterTile.prtEncoder)
   implicit val tripEnc = Encoders.tuple(ProjectedRasterTile.prtEncoder, ProjectedRasterTile.prtEncoder, ProjectedRasterTile.prtEncoder)

core/src/test/scala/org/locationtech/rasterframes/encoders/CatalystSerializerSpec.scala

@@ -105,7 +105,8 @@ class CatalystSerializerSpec extends TestEnvironment with TestData {
     it("should serialize RasterRef") {
       // TODO: Decide if RasterRef should be encoded 'flat', non-'flat', or depends
       val src = RasterSource(remoteCOGSingleband1)
-      val value = RasterRef(src, 0, Some(src.extent.buffer(-3.0)))
+      val ext = src.extent.buffer(-3.0)
+      val value = RasterRef(src, 0, Some(ext), Some(src.rasterExtent.gridBoundsFor(ext)))
       assertConsistent(value)
       assertInvertable(value)
     }

core/src/test/scala/org/locationtech/rasterframes/ref/RasterRefSpec.scala

@@ -47,9 +47,9 @@ class RasterRefSpec extends TestEnvironment with TestData {
 
   trait Fixture {
     val src = RasterSource(remoteCOGSingleband1)
-    val fullRaster = RasterRef(src, 0, None)
+    val fullRaster = RasterRef(src, 0, None, None)
     val subExtent = sub(src.extent)
-    val subRaster = RasterRef(src, 0, Some(subExtent))
+    val subRaster = RasterRef(src, 0, Some(subExtent), Some(src.rasterExtent.gridBoundsFor(subExtent)))
   }
 
   import spark.implicits._
@@ -171,7 +171,7 @@ class RasterRefSpec extends TestEnvironment with TestData {
       val src = RasterSource(remoteMODIS)
       val dims = src
         .layoutExtents(NOMINAL_TILE_DIMS)
-        .map(e => RasterRef(src, 0, Some(e)))
+        .map(e => RasterRef(src, 0, Some(e), None))
         .map(_.dimensions)
         .distinct