Merge remote-tracking branch 'locationtech/develop' into feature/nodata-add-doc

Author: vpipkt

README.md

@@ -1,10 +1,10 @@
-<img src="docs/src/main/paradox/_template/images/RasterFramesLogo.png" width="300px"/><sup style="vertical-align: top;">&trade;</sup>
+<img src="docs/src/main/paradox/_template/images/RasterFramesLogo.png" width="300px"/><sup style="vertical-align: top;">&reg;</sup>
 
  [![Join the chat at https://gitter.im/s22s/raster-frames](https://badges.gitter.im/s22s/raster-frames.svg)](https://gitter.im/s22s/raster-frames?utm_source=badge&utm_medium=badge&utm_campaign=pr-badge&utm_content=badge)
 
 _RasterFrames™_ brings the power of Spark DataFrames to geospatial raster data, empowered by the map algebra and tile layer operations of [GeoTrellis](https://geotrellis.io/).
 
-<img src="docs/src/main/paradox/RasterFramePipeline.svg" width="600px"/>
+<img src="docs/src/main/paradox/RasterFramePipeline.png" width="600px"/>
 
 Please see the [Getting Started](http://rasterframes.io/getting-started.html) section of the Users' Manual to start using RasterFrames.
 
@@ -54,6 +54,6 @@ The `pyrasterframes` build instructions are located at [pyrasterframes/src/main/
 
 ## Copyright and License
 
-RasterFrames is released under the Apache 2.0 License, copyright Astraea, Inc. 2017-2018.
+RasterFrames is released under the Apache 2.0 License, copyright Astraea, Inc. 2017-2019.
 
 

core/src/it/scala/org/locationtech/rasterframes/ref/RasterSourceIT.scala

@@ -63,8 +63,8 @@ class RasterSourceIT extends TestEnvironment with TestData {
     }
   }
 
-  if (RasterSource.IsGDAL.hasGDAL) {
-    println("GDAL version: " + GDALWarp.get_version_info("--version"))
+  if (GDALRasterSource.hasGDAL) {
+    println("GDAL version: " + GDALRasterSource.gdalVersion())
 
     describe("GDAL support") {
 

core/src/main/resources/reference.conf

@@ -12,7 +12,7 @@ vlm.gdal {
       //CPL_DEBUG = "OFF"
       AWS_REQUEST_PAYER = "requester"
       GDAL_DISABLE_READDIR_ON_OPEN = "YES"
-      CPL_VSIL_CURL_ALLOWED_EXTENSIONS = ".tif,.tiff,.jp2,.mrf,.idx,.lrc,.mrf.aux.xml"
+      CPL_VSIL_CURL_ALLOWED_EXTENSIONS = ".tif,.tiff,.jp2,.mrf,.idx,.lrc,.mrf.aux.xml,.vrt"
     }
     // set this to `false` if CPL_DEBUG is `ON`
     useExceptions = true

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

@@ -154,4 +154,9 @@ object CatalystSerializer extends StandardSerializers {
   implicit class WithFromRow(val r: Row) extends AnyVal {
     def to[T >: Null: CatalystSerializer]: T = if (r == null) null else CatalystSerializer[T].fromRow(r)
   }
+
+  implicit class WithTypeConformity(val left: DataType) extends AnyVal {
+    def conformsTo[T >: Null: CatalystSerializer]: Boolean =
+      org.apache.spark.sql.rf.WithTypeConformity(left).conformsTo(schemaOf[T])
+  }
 }

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

@@ -24,14 +24,14 @@ package org.locationtech.rasterframes.encoders
 import java.net.URI
 import java.sql.Timestamp
 
-import org.locationtech.rasterframes.model._
 import org.locationtech.rasterframes.stats.{CellHistogram, CellStatistics, LocalCellStatistics}
 import org.locationtech.jts.geom.Envelope
 import geotrellis.proj4.CRS
 import geotrellis.raster.{CellSize, CellType, Raster, Tile, TileLayout}
 import geotrellis.spark.tiling.LayoutDefinition
 import geotrellis.spark.{KeyBounds, SpaceTimeKey, SpatialKey, TemporalKey, TemporalProjectedExtent, TileLayerMetadata}
 import geotrellis.vector.{Extent, ProjectedExtent}
+import org.apache.spark.sql.{Encoder, Encoders}
 import org.apache.spark.sql.catalyst.encoders.ExpressionEncoder
 import org.locationtech.geomesa.spark.jts.encoders.SpatialEncoders
 import org.locationtech.rasterframes.model.{CellContext, Cells, TileContext, TileDataContext}
@@ -70,6 +70,7 @@ trait StandardEncoders extends SpatialEncoders {
   implicit def cellsEncoder: ExpressionEncoder[Cells] = Cells.encoder
   implicit def tileContextEncoder: ExpressionEncoder[TileContext] = TileContext.encoder
   implicit def tileDataContextEncoder: ExpressionEncoder[TileDataContext] = TileDataContext.encoder
+  implicit def extentTilePairEncoder: Encoder[(ProjectedExtent, Tile)] = Encoders.tuple(projectedExtentEncoder, singlebandTileEncoder)
 
 
 }

core/src/main/scala/org/locationtech/rasterframes/expressions/DynamicExtractors.scala

@@ -21,13 +21,15 @@
 
 package org.locationtech.rasterframes.expressions
 
+import geotrellis.proj4.CRS
 import geotrellis.raster.{CellGrid, Tile}
 import org.apache.spark.sql.Row
 import org.apache.spark.sql.catalyst.InternalRow
-import org.apache.spark.sql.rf.{TileUDT, _}
+import org.apache.spark.sql.rf.{RasterSourceUDT, TileUDT}
 import org.apache.spark.sql.types._
+import org.apache.spark.unsafe.types.UTF8String
 import org.locationtech.rasterframes.encoders.CatalystSerializer._
-import org.locationtech.rasterframes.model.TileContext
+import org.locationtech.rasterframes.model.{LazyCRS, TileContext}
 import org.locationtech.rasterframes.ref.{ProjectedRasterLike, RasterRef, RasterSource}
 import org.locationtech.rasterframes.tiles.ProjectedRasterTile
 
@@ -38,7 +40,7 @@ object DynamicExtractors {
     case _: TileUDT =>
       (row: InternalRow) =>
         (row.to[Tile](TileUDT.tileSerializer), None)
-    case t if t.conformsTo(schemaOf[ProjectedRasterTile]) =>
+    case t if t.conformsTo[ProjectedRasterTile] =>
       (row: InternalRow) => {
         val prt = row.to[ProjectedRasterTile]
         (prt, Some(TileContext(prt)))
@@ -48,7 +50,7 @@ object DynamicExtractors {
   lazy val rowTileExtractor: PartialFunction[DataType, Row => (Tile, Option[TileContext])] = {
     case _: TileUDT =>
       (row: Row) =>  (row.to[Tile](TileUDT.tileSerializer), None)
-    case t if t.conformsTo(schemaOf[ProjectedRasterTile]) =>
+    case t if t.conformsTo[ProjectedRasterTile] =>
       (row: Row) => {
         val prt = row.to[ProjectedRasterTile]
         (prt, Some(TileContext(prt)))
@@ -59,9 +61,9 @@ object DynamicExtractors {
   lazy val projectedRasterLikeExtractor: PartialFunction[DataType, InternalRow ⇒ ProjectedRasterLike] = {
     case _: RasterSourceUDT ⇒
       (row: InternalRow) => row.to[RasterSource](RasterSourceUDT.rasterSourceSerializer)
-    case t if t.conformsTo(schemaOf[ProjectedRasterTile]) =>
+    case t if t.conformsTo[ProjectedRasterTile] =>
       (row: InternalRow) => row.to[ProjectedRasterTile]
-    case t if t.conformsTo(schemaOf[RasterRef]) =>
+    case t if t.conformsTo[RasterRef] =>
       (row: InternalRow) => row.to[RasterRef]
   }
 
@@ -71,12 +73,19 @@ object DynamicExtractors {
       (row: InternalRow) => row.to[Tile](TileUDT.tileSerializer)
     case _: RasterSourceUDT =>
       (row: InternalRow) => row.to[RasterSource](RasterSourceUDT.rasterSourceSerializer)
-    case t if t.conformsTo(schemaOf[RasterRef]) ⇒
+    case t if t.conformsTo[RasterRef] ⇒
       (row: InternalRow) => row.to[RasterRef]
-    case t if t.conformsTo(schemaOf[ProjectedRasterTile]) =>
+    case t if t.conformsTo[ProjectedRasterTile] =>
       (row: InternalRow) => row.to[ProjectedRasterTile]
   }
 
+  lazy val crsExtractor: PartialFunction[DataType, Any => CRS] = {
+    case _: StringType =>
+      (v: Any) => LazyCRS(v.asInstanceOf[UTF8String].toString)
+    case t if t.conformsTo[CRS] =>
+      (v: Any) => v.asInstanceOf[InternalRow].to[CRS]
+  }
+
   sealed trait TileOrNumberArg
   sealed trait NumberArg extends TileOrNumberArg
   case class TileArg(tile: Tile, ctx: Option[TileContext]) extends TileOrNumberArg
@@ -113,4 +122,5 @@ object DynamicExtractors {
       case c: Char  => IntegerArg(c.toInt)
     }
   }
+
 }

core/src/main/scala/org/locationtech/rasterframes/expressions/SpatialRelation.scala

@@ -30,7 +30,6 @@ import org.apache.spark.sql.catalyst.encoders.ExpressionEncoder
 import org.apache.spark.sql.catalyst.expressions.codegen.CodegenFallback
 import org.apache.spark.sql.catalyst.expressions.{ScalaUDF, _}
 import org.apache.spark.sql.jts.AbstractGeometryUDT
-import org.apache.spark.sql.rf.WithTypeConformity
 import org.apache.spark.sql.types._
 import org.locationtech.geomesa.spark.jts.udf.SpatialRelationFunctions._
 
@@ -48,7 +47,7 @@ abstract class SpatialRelation extends BinaryExpression
       case r: InternalRow ⇒
         expr.dataType match {
           case udt: AbstractGeometryUDT[_] ⇒ udt.deserialize(r)
-          case dt if dt.conformsTo(schemaOf[Extent]) =>
+          case dt if dt.conformsTo[Extent] =>
             val extent = r.to[Extent]
             extent.jtsGeom
         }

core/src/main/scala/org/locationtech/rasterframes/expressions/aggregates/LocalMeanAggregate.scala

@@ -22,7 +22,7 @@
 package org.locationtech.rasterframes.expressions.aggregates
 
 import org.locationtech.rasterframes.expressions.UnaryRasterAggregate
-import org.locationtech.rasterframes.expressions.localops.{Add => AddTiles, Divide => DivideTiles}
+import org.locationtech.rasterframes.expressions.localops.{BiasedAdd, Divide => DivideTiles}
 import org.locationtech.rasterframes.expressions.transformers.SetCellType
 import geotrellis.raster.Tile
 import geotrellis.raster.mapalgebra.local
@@ -59,16 +59,16 @@ case class LocalMeanAggregate(child: Expression) extends UnaryRasterAggregate {
   override lazy val updateExpressions: Seq[Expression] = Seq(
     If(IsNull(count),
       SetCellType(Defined(child), Literal("int32")),
-      If(IsNull(child), count, AddTiles(count, Defined(child)))
+      If(IsNull(child), count, BiasedAdd(count, Defined(child)))
     ),
     If(IsNull(sum),
       SetCellType(child, Literal("float64")),
-      If(IsNull(child), sum, AddTiles(sum, child))
+      If(IsNull(child), sum, BiasedAdd(sum, child))
     )
   )
   override val mergeExpressions: Seq[Expression] = Seq(
-    AddTiles(count.left, count.right),
-    AddTiles(sum.left, sum.right)
+    BiasedAdd(count.left, count.right),
+    BiasedAdd(sum.left, sum.right)
   )
   override lazy val evaluateExpression: Expression = DivideTiles(sum, count)
 }

core/src/main/scala/org/locationtech/rasterframes/expressions/aggregates/ProjectedLayerMetadataAggregate.scala

@@ -0,0 +1,179 @@
+/*
+ * 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 org.locationtech.rasterframes.expressions.aggregates
+
+import org.locationtech.rasterframes._
+import org.locationtech.rasterframes.encoders.CatalystSerializer
+import org.locationtech.rasterframes.encoders.CatalystSerializer._
+import org.locationtech.rasterframes.model.TileDimensions
+import geotrellis.proj4.{CRS, Transform}
+import geotrellis.raster._
+import geotrellis.raster.reproject.{Reproject, ReprojectRasterExtent}
+import geotrellis.spark.tiling.LayoutDefinition
+import geotrellis.spark.{KeyBounds, SpatialKey, TileLayerMetadata}
+import geotrellis.vector.Extent
+import org.apache.spark.sql.expressions.{MutableAggregationBuffer, UserDefinedAggregateFunction}
+import org.apache.spark.sql.types.{DataType, StructField, StructType}
+import org.apache.spark.sql.{Column, Row, TypedColumn}
+
+class ProjectedLayerMetadataAggregate(destCRS: CRS, destDims: TileDimensions) extends UserDefinedAggregateFunction {
+  import ProjectedLayerMetadataAggregate._
+
+  override def inputSchema: StructType = CatalystSerializer[InputRecord].schema
+
+  override def bufferSchema: StructType = CatalystSerializer[BufferRecord].schema
+
+  override def dataType: DataType = CatalystSerializer[TileLayerMetadata[SpatialKey]].schema
+
+  override def deterministic: Boolean = true
+
+  override def initialize(buffer: MutableAggregationBuffer): Unit = ()
+
+  override def update(buffer: MutableAggregationBuffer, input: Row): Unit = {
+    if(!input.isNullAt(0)) {
+      val in = input.to[InputRecord]
+
+      if(buffer.isNullAt(0)) {
+        in.toBufferRecord(destCRS).write(buffer)
+      }
+      else {
+        val br = buffer.to[BufferRecord]
+        br.merge(in.toBufferRecord(destCRS)).write(buffer)
+      }
+    }
+  }
+
+  override def merge(buffer1: MutableAggregationBuffer, buffer2: Row): Unit = {
+    (buffer1.isNullAt(0), buffer2.isNullAt(0)) match {
+      case (false, false) ⇒
+        val left = buffer1.to[BufferRecord]
+        val right = buffer2.to[BufferRecord]
+        left.merge(right).write(buffer1)
+      case (true, false) ⇒ buffer2.to[BufferRecord].write(buffer1)
+      case _ ⇒ ()
+    }
+  }
+
+  override def evaluate(buffer: Row): Any = {
+    import org.locationtech.rasterframes.encoders.CatalystSerializer._
+    val buf = buffer.to[BufferRecord]
+
+    val re = RasterExtent(buf.extent, buf.cellSize)
+    val layout = LayoutDefinition(re, destDims.cols, destDims.rows)
+
+    val kb = KeyBounds(layout.mapTransform(buf.extent))
+    TileLayerMetadata(buf.cellType, layout, buf.extent, destCRS, kb).toRow
+  }
+}
+
+object ProjectedLayerMetadataAggregate {
+  import org.locationtech.rasterframes.encoders.StandardEncoders._
+
+  /** Primary user facing constructor */
+  def apply(destCRS: CRS, extent: Column, crs: Column, cellType: Column,  tileSize: Column): TypedColumn[Any, TileLayerMetadata[SpatialKey]] =
+  // Ordering must match InputRecord schema
+    new ProjectedLayerMetadataAggregate(destCRS, TileDimensions(NOMINAL_TILE_SIZE, NOMINAL_TILE_SIZE))(extent, crs, cellType, tileSize).as[TileLayerMetadata[SpatialKey]]
+
+  def apply(destCRS: CRS, destDims: TileDimensions, extent: Column, crs: Column, cellType: Column,  tileSize: Column): TypedColumn[Any, TileLayerMetadata[SpatialKey]] =
+  // Ordering must match InputRecord schema
+    new ProjectedLayerMetadataAggregate(destCRS, destDims)(extent, crs, cellType, tileSize).as[TileLayerMetadata[SpatialKey]]
+
+  private[expressions]
+  case class InputRecord(extent: Extent, crs: CRS, cellType: CellType, tileSize: TileDimensions) {
+    def toBufferRecord(destCRS: CRS): BufferRecord = {
+      val transform = Transform(crs, destCRS)
+
+      val re = ReprojectRasterExtent(
+        RasterExtent(extent, tileSize.cols, tileSize.rows),
+        transform, Reproject.Options.DEFAULT
+      )
+
+      BufferRecord(
+        re.extent,
+        cellType,
+        re.cellSize
+      )
+    }
+  }
+
+  private[expressions]
+  object InputRecord {
+    implicit val serializer: CatalystSerializer[InputRecord] = new CatalystSerializer[InputRecord]{
+      override def schema: StructType = StructType(Seq(
+        StructField("extent", CatalystSerializer[Extent].schema, false),
+        StructField("crs", CatalystSerializer[CRS].schema, false),
+        StructField("cellType", CatalystSerializer[CellType].schema, false),
+        StructField("tileSize", CatalystSerializer[TileDimensions].schema, false)
+      ))
+
+      override protected def to[R](t: InputRecord, io: CatalystIO[R]): R =
+        throw new IllegalStateException("InputRecord is input only.")
+
+      override protected def from[R](t: R, io: CatalystIO[R]): InputRecord = InputRecord(
+        io.get[Extent](t, 0),
+        io.get[CRS](t, 1),
+        io.get[CellType](t, 2),
+        io.get[TileDimensions](t, 3)
+      )
+    }
+  }
+
+  private[expressions]
+  case class BufferRecord(extent: Extent, cellType: CellType, cellSize: CellSize) {
+    def merge(that: BufferRecord): BufferRecord = {
+      val ext = this.extent.combine(that.extent)
+      val ct = this.cellType.union(that.cellType)
+      val cs = if (this.cellSize.resolution < that.cellSize.resolution) this.cellSize else that.cellSize
+      BufferRecord(ext, ct, cs)
+    }
+
+    def write(buffer: MutableAggregationBuffer): Unit = {
+      val encoded = (this).toRow
+      for(i <- 0 until encoded.size) {
+        buffer(i) = encoded(i)
+      }
+    }
+  }
+
+  private[expressions]
+  object BufferRecord {
+    implicit val serializer: CatalystSerializer[BufferRecord] = new CatalystSerializer[BufferRecord] {
+      override def schema: StructType = StructType(Seq(
+        StructField("extent", CatalystSerializer[Extent].schema, true),
+        StructField("cellType", CatalystSerializer[CellType].schema, true),
+        StructField("cellSize", CatalystSerializer[CellSize].schema, true)
+      ))
+
+      override protected def to[R](t: BufferRecord, io: CatalystIO[R]): R = io.create(
+        io.to(t.extent),
+        io.to(t.cellType),
+        io.to(t.cellSize)
+      )
+
+      override protected def from[R](t: R, io: CatalystIO[R]): BufferRecord = BufferRecord(
+        io.get[Extent](t, 0),
+        io.get[CellType](t, 1),
+        io.get[CellSize](t, 2)
+      )
+    }
+  }
+}