Merge pull request #373 from s22s/fix/372

Fixed eager loading of RasterRef fields.

Author: metasim

core/src/main/scala/org/apache/spark/sql/rf/TileUDT.scala

@@ -26,6 +26,7 @@ import org.apache.spark.sql.types.{DataType, _}
 import org.locationtech.rasterframes.encoders.CatalystSerializer
 import org.locationtech.rasterframes.encoders.CatalystSerializer._
 import org.locationtech.rasterframes.model.{Cells, TileDataContext}
+import org.locationtech.rasterframes.ref.RasterRef.RasterRefTile
 import org.locationtech.rasterframes.tiles.InternalRowTile
 
 
@@ -75,12 +76,15 @@ case object TileUDT  {
   implicit def tileSerializer: CatalystSerializer[Tile] = new CatalystSerializer[Tile] {
 
     override val schema: StructType = StructType(Seq(
-      StructField("cell_context", schemaOf[TileDataContext], false),
+      StructField("cell_context", schemaOf[TileDataContext], true),
       StructField("cell_data", schemaOf[Cells], false)
     ))
 
     override def to[R](t: Tile, io: CatalystIO[R]): R = io.create(
-      io.to(TileDataContext(t)),
+      t match {
+        case _: RasterRefTile => null
+        case o => io.to(TileDataContext(o))
+      },
       io.to(Cells(t))
     )
 

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

@@ -43,7 +43,7 @@ abstract class DelegatingRasterSource(source: URI, delegateBuilder: () => GTRast
       f(_delRef)
     }
     catch {
-      // On this exeception we attempt to recreate the delegate and read again.
+      // On this Exeception we attempt to recreate the delegate and read again.
       case _: java.nio.BufferUnderflowException =>
         _delRef = null
         val newDel = delegateBuilder()
@@ -62,7 +62,7 @@ abstract class DelegatingRasterSource(source: URI, delegateBuilder: () => GTRast
   override def hashCode(): Int = source.hashCode()
 
   // This helps reduce header reads between serializations
-  def info: SimpleRasterInfo = SimpleRasterInfo.cache.get(source.toASCIIString, _ =>
+  def info: SimpleRasterInfo = SimpleRasterInfo(source.toASCIIString, _ =>
     retryableRead(rs => SimpleRasterInfo(rs))
   )
 

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

@@ -48,7 +48,7 @@ case class GDALRasterSource(source: URI) extends RasterSource with URIRasterSour
     VLMRasterSource(tweaked)
   }
 
-  protected def tiffInfo = SimpleRasterInfo.cache.get(source.toASCIIString, _ => SimpleRasterInfo(gdal))
+  protected def tiffInfo = SimpleRasterInfo(source.toASCIIString, _ => SimpleRasterInfo(gdal))
 
   override def crs: CRS = tiffInfo.crs
 

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

@@ -61,12 +61,12 @@ object RasterRef extends LazyLogging {
   private val log = logger
 
   case class RasterRefTile(rr: RasterRef) extends ProjectedRasterTile {
-    val extent: Extent = rr.extent
-    val crs: CRS = rr.crs
+    def extent: Extent = rr.extent
+    def crs: CRS = rr.crs
     override def cellType = rr.cellType
 
-    override val cols: Int = rr.cols
-    override val rows: Int = rr.rows
+    override def cols: Int = rr.cols
+    override def rows: Int = rr.rows
 
     protected def delegate: Tile = rr.realizedTile
     // NB: This saves us from stack overflow exception

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

@@ -92,7 +92,7 @@ object RasterSource extends LazyLogging {
 
   val cacheTimeout: Duration = Duration.fromNanos(rfConfig.getDuration("raster-source-cache-timeout").toNanos)
 
-  private val rsCache = Scaffeine()
+  private[ref] val rsCache = Scaffeine()
     .recordStats()
     .expireAfterAccess(RasterSource.cacheTimeout)
     .build[String, RasterSource]

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

@@ -44,6 +44,11 @@ case class SimpleRasterInfo(
 )
 
 object SimpleRasterInfo {
+  // Not a fan of this.... need a better abstraction that doesn't put the
+  // type-specific logic in here.
+  def apply(key: String, builder: String => SimpleRasterInfo): SimpleRasterInfo =
+    cache.get(key, builder)
+
   def apply(info: GeoTiffReader.GeoTiffInfo): SimpleRasterInfo =
     SimpleRasterInfo(
       info.segmentLayout.totalCols,
@@ -76,10 +81,9 @@ object SimpleRasterInfo {
     )
   }
 
-  private[rasterframes]
-  lazy val cache = Scaffeine()
+  private lazy val cache = Scaffeine()
     .recordStats()
     .build[String, SimpleRasterInfo]
 
   def cacheStats = cache.stats()
-}
+}

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

@@ -72,12 +72,15 @@ object ProjectedRasterTile {
   }
   implicit val serializer: CatalystSerializer[ProjectedRasterTile] = new CatalystSerializer[ProjectedRasterTile] {
     override val schema: StructType = StructType(Seq(
-      StructField("tile_context", schemaOf[TileContext], false),
+      StructField("tile_context", schemaOf[TileContext], true),
       StructField("tile", TileType, false))
     )
 
     override protected def to[R](t: ProjectedRasterTile, io: CatalystIO[R]): R = io.create(
-      io.to(TileContext(t.extent, t.crs)),
+      t match {
+        case _: RasterRefTile => null
+        case o => io.to(TileContext(o.extent, o.crs))
+      },
       io.to[Tile](t)(TileUDT.tileSerializer)
     )
 

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

@@ -232,23 +232,6 @@ class RasterFrameSpec extends TestEnvironment with MetadataKeys
       assert(bounds._2 === SpaceTimeKey(3, 1, now))
     }
 
-//    it("should clip TileLayerMetadata extent") {
-//      val tiled = sampleTileLayerRDD
-//
-//      val rf = tiled.reproject(LatLng, tiled.metadata.layout)._2.toLayer
-//
-//      val worldish = Extent(-179, -89, 179, 89)
-//      val areaish = Extent(-90, 30, -81, 40)
-//
-//      val orig = rf.tileLayerMetadata.widen.extent
-//      assert(worldish.contains(orig))
-//      assert(areaish.contains(orig))
-//
-//      val clipped = rf.clipLayerExtent.tileLayerMetadata.widen.extent
-//      assert(!clipped.contains(worldish))
-//      assert(clipped.contains(areaish))
-//    }
-
     def basicallySame(expected: Extent, computed: Extent): Unit = {
       val components = Seq(
         (expected.xmin, computed.xmin),

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

@@ -252,5 +252,16 @@ class RasterRefSpec extends TestEnvironment with TestData {
         assertEqual(t.toArrayTile(), result)
       }
     }
+
+    it("should construct a RasterRefTile without I/O") {
+      new Fixture {
+        // SimpleRasterInfo is a proxy for header data requests.
+        val start = SimpleRasterInfo.cacheStats.hitCount()
+        val t: ProjectedRasterTile = RasterRefTile(subRaster)
+        val result = Seq(t).toDF("tile").first()
+        val end = SimpleRasterInfo.cacheStats.hitCount()
+        end should be(start)
+      }
+    }
   }
 }

project/plugins.sbt

@@ -15,5 +15,6 @@ addSbtPlugin("com.eed3si9n" % "sbt-unidoc" % "0.4.1")
 addSbtPlugin("net.vonbuchholtz" % "sbt-dependency-check" % "0.2.10")
 addSbtPlugin("com.github.gseitz" %% "sbt-release" % "1.0.9")
 addSbtPlugin("com.typesafe.sbt" % "sbt-native-packager" % "1.3.19")
+addSbtPlugin("org.scoverage" % "sbt-scoverage" % "1.6.0")