Merge remote-tracking branch 'locationtech/develop' into feature/tile-quantile

Author: vpipkt

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

@@ -30,7 +30,7 @@ import org.locationtech.rasterframes.expressions.aggregates.TileRasterizerAggreg
 
 class RasterRefIT extends TestEnvironment {
   describe("practical subregion reads") {
-    ignore("should construct a natural color composite") {
+    it("should construct a natural color composite") {
       import spark.implicits._
       def scene(idx: Int) = URI.create(s"https://landsat-pds.s3.us-west-2.amazonaws.com" +
         s"/c1/L8/176/039/LC08_L1TP_176039_20190703_20190718_01_T1/LC08_L1TP_176039_20190703_20190718_01_T1_B$idx.TIF")
@@ -55,11 +55,11 @@ class RasterRefIT extends TestEnvironment {
         stats.get.dataCells shouldBe > (1000L)
       }
 
-      //import geotrellis.raster.io.geotiff.{GeoTiffOptions, MultibandGeoTiff, Tiled}
-      //import geotrellis.raster.io.geotiff.compression.{DeflateCompression, NoCompression}
-      //import geotrellis.raster.io.geotiff.tags.codes.ColorSpace
-      //val tiffOptions = GeoTiffOptions(Tiled,  DeflateCompression, ColorSpace.RGB)
-      //MultibandGeoTiff(raster, raster.crs, tiffOptions).write("target/composite.tif")
+      import geotrellis.raster.io.geotiff.compression.DeflateCompression
+      import geotrellis.raster.io.geotiff.tags.codes.ColorSpace
+      import geotrellis.raster.io.geotiff.{GeoTiffOptions, MultibandGeoTiff, Tiled}
+      val tiffOptions = GeoTiffOptions(Tiled,  DeflateCompression, ColorSpace.RGB)
+      MultibandGeoTiff(raster.raster, raster.crs, tiffOptions).write("target/composite.tif")
     }
   }
 }

core/src/main/resources/reference.conf

@@ -17,6 +17,8 @@ vlm.gdal {
       GDAL_CACHEMAX = 512
       GDAL_PAM_ENABLED = "NO"
       CPL_VSIL_CURL_CHUNK_SIZE = 1000000
+      GDAL_HTTP_MAX_RETRY=4
+      GDAL_HTTP_RETRY_DELAY=1
     }
     // set this to `false` if CPL_DEBUG is `ON`
     useExceptions = true

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

@@ -38,7 +38,7 @@ import org.locationtech.rasterframes.util.KryoSupport
 @SQLUserDefinedType(udt = classOf[RasterSourceUDT])
 class RasterSourceUDT extends UserDefinedType[RasterSource] {
   import RasterSourceUDT._
-  override def typeName = "rf_rastersource"
+  override def typeName = "rastersource"
 
   override def pyUDT: String = "pyrasterframes.rf_types.RasterSourceUDT"
 
@@ -58,7 +58,6 @@ class RasterSourceUDT extends UserDefinedType[RasterSource] {
       }
       .orNull
 
-
   private[sql] override def acceptsType(dataType: DataType) = dataType match {
     case _: RasterSourceUDT ⇒ true
     case _ ⇒ super.acceptsType(dataType)

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

@@ -73,7 +73,7 @@ case object TileUDT  {
 
   final val typeName: String = "tile"
 
-  implicit def tileSerializer: CatalystSerializer[Tile] = new CatalystSerializer[Tile] {
+  implicit val tileSerializer: CatalystSerializer[Tile] = new CatalystSerializer[Tile] {
 
     override val schema: StructType = StructType(Seq(
       StructField("cell_context", schemaOf[TileDataContext], true),

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

@@ -22,7 +22,7 @@
 package org.locationtech.rasterframes.expressions
 
 import geotrellis.proj4.CRS
-import geotrellis.raster.{CellGrid, Tile}
+import geotrellis.raster.{CellGrid, Raster, Tile}
 import geotrellis.vector.Extent
 import org.apache.spark.sql.Row
 import org.apache.spark.sql.catalyst.InternalRow
@@ -61,6 +61,11 @@ 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[Raster[Tile]] =>
+      (row: Row) => {
+        val rt = row.to[Raster[Tile]]
+        (rt.tile, None)
+      }
     case t if t.conformsTo[ProjectedRasterTile] =>
       (row: Row) => {
         val prt = row.to[ProjectedRasterTile]

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

@@ -107,7 +107,7 @@ object HistogramAggregate {
 
   /** Adapter hack to allow UserDefinedAggregateFunction to be referenced as an expression. */
   @ExpressionDescription(
-    usage = "_FUNC_(tile) - Compute aggregate cell histogram over a tile column.",
+    usage = "_FUNC_(tile) - Compute aggregate cell histogram over fa tile column.",
     arguments = """
   Arguments:
     * tile - tile column to analyze""",

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

@@ -23,18 +23,18 @@ package org.locationtech.rasterframes.expressions.aggregates
 
 import geotrellis.proj4.CRS
 import geotrellis.raster.reproject.Reproject
-import geotrellis.raster.resample.ResampleMethod
-import geotrellis.raster.{ArrayTile, CellType, MultibandTile, ProjectedRaster, Raster, Tile}
+import geotrellis.raster.resample.{Bilinear, ResampleMethod}
+import geotrellis.raster.{ArrayTile, CellType, MultibandTile, ProjectedRaster, Tile}
 import geotrellis.spark.{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, DataFrame, Row, TypedColumn}
 import org.locationtech.rasterframes._
-import org.locationtech.rasterframes.util._
 import org.locationtech.rasterframes.encoders.CatalystSerializer._
 import org.locationtech.rasterframes.expressions.aggregates.TileRasterizerAggregate.ProjectedRasterDefinition
 import org.locationtech.rasterframes.model.TileDimensions
+import org.locationtech.rasterframes.util._
 import org.slf4j.LoggerFactory
 
 /**
@@ -58,22 +58,22 @@ class TileRasterizerAggregate(prd: ProjectedRasterDefinition) extends UserDefine
     StructField("tile_buffer", TileType)
   ))
 
-  override def dataType: DataType = schemaOf[Raster[Tile]]
+  override def dataType: DataType = TileType
 
   override def initialize(buffer: MutableAggregationBuffer): Unit = {
-    buffer(0) = ArrayTile.empty(prd.cellType, prd.totalCols, prd.totalRows)
+    buffer(0) = ArrayTile.empty(prd.destinationCellType, prd.totalCols, prd.totalRows)
   }
 
   override def update(buffer: MutableAggregationBuffer, input: Row): Unit = {
     val crs = input.getAs[Row](0).to[CRS]
     val extent = input.getAs[Row](1).to[Extent]
 
-    val localExtent = extent.reproject(crs, prd.crs)
+    val localExtent = extent.reproject(crs, prd.destinationCRS)
 
-    if (prd.extent.intersects(localExtent)) {
-      val localTile = input.getAs[Tile](2).reproject(extent, crs, prd.crs, projOpts)
+    if (prd.destinationExtent.intersects(localExtent)) {
+      val localTile = input.getAs[Tile](2).reproject(extent, crs, prd.destinationCRS, projOpts)
       val bt = buffer.getAs[Tile](0)
-      val merged = bt.merge(prd.extent, localExtent, localTile.tile, prd.sampler)
+      val merged = bt.merge(prd.destinationExtent, localExtent, localTile.tile, prd.sampler)
       buffer(0) = merged
     }
   }
@@ -84,39 +84,36 @@ class TileRasterizerAggregate(prd: ProjectedRasterDefinition) extends UserDefine
     buffer1(0) = leftTile.merge(rightTile)
   }
 
-  override def evaluate(buffer: Row): Raster[Tile] = {
-    val t = buffer.getAs[Tile](0)
-    Raster(t, prd.extent)
-  }
+  override def evaluate(buffer: Row): Tile = buffer.getAs[Tile](0)
 }
 
 object TileRasterizerAggregate {
-  val nodeName = "rf_agg_raster"
-  /**  Convenience grouping of  parameters needed for running aggregate. */
-  case class ProjectedRasterDefinition(totalCols: Int, totalRows: Int, cellType: CellType, crs: CRS, extent: Extent, sampler: ResampleMethod = ResampleMethod.DEFAULT)
+  @transient
+  private lazy val logger = LoggerFactory.getLogger(getClass)
 
-  object ProjectedRasterDefinition {
-    def apply(tlm: TileLayerMetadata[_]): ProjectedRasterDefinition = apply(tlm, ResampleMethod.DEFAULT)
+  /** Convenience grouping of  parameters needed for running aggregate. */
+  case class ProjectedRasterDefinition(totalCols: Int, totalRows: Int, destinationCellType: CellType, destinationCRS: CRS,
+    destinationExtent: Extent, sampler: ResampleMethod)
 
+  object ProjectedRasterDefinition {
     def apply(tlm: TileLayerMetadata[_], sampler: ResampleMethod): ProjectedRasterDefinition = {
       // Try to determine the actual dimensions of our data coverage
       val TileDimensions(cols, rows) = tlm.totalDimensions
       new ProjectedRasterDefinition(cols, rows, tlm.cellType, tlm.crs, tlm.extent, sampler)
     }
   }
 
-  @transient
-  private lazy val logger = LoggerFactory.getLogger(getClass)
-
-  def apply(prd: ProjectedRasterDefinition, crsCol: Column, extentCol: Column, tileCol: Column): TypedColumn[Any, Raster[Tile]] = {
-
+  def apply(prd: ProjectedRasterDefinition, crsCol: Column, extentCol: Column, tileCol: Column): TypedColumn[Any, Tile] = {
     if (prd.totalCols.toDouble * prd.totalRows * 64.0 > Runtime.getRuntime.totalMemory() * 0.5)
       logger.warn(
         s"You've asked for the construction of a very large image (${prd.totalCols} x ${prd.totalRows}). Out of memory error likely.")
 
-    new TileRasterizerAggregate(prd)(crsCol, extentCol, tileCol).as(nodeName).as[Raster[Tile]]
+    new TileRasterizerAggregate(prd)(crsCol, extentCol, tileCol)
+      .as("rf_agg_overview_raster")
+      .as[Tile]
   }
 
+  /** Extract a multiband raster from all tile columns. */
   def collect(df: DataFrame, destCRS: CRS, destExtent: Option[Extent], rasterDims: Option[TileDimensions]): ProjectedRaster[MultibandTile] = {
     val tileCols = WithDataFrameMethods(df).tileColumns
     require(tileCols.nonEmpty, "need at least one tile column")
@@ -149,7 +146,7 @@ object TileRasterizerAggregate {
       .first()
     logger.debug(s"Collected TileLayerMetadata: ${tlm.toString}")
 
-    val c = ProjectedRasterDefinition(tlm)
+    val c = ProjectedRasterDefinition(tlm, Bilinear)
 
     val config = rasterDims
       .map { dims =>
@@ -158,11 +155,11 @@ object TileRasterizerAggregate {
       .getOrElse(c)
 
     destExtent.map { ext =>
-      c.copy(extent = ext)
+      c.copy(destinationExtent = ext)
     }
 
     val aggs = tileCols
-      .map(t => TileRasterizerAggregate(config, crsCol, extCol, rf_tile(t))("tile").as(t.columnName))
+      .map(t => TileRasterizerAggregate(config, crsCol, extCol, rf_tile(t)).as(t.columnName))
 
     val agg = df.select(aggs: _*)
 

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

@@ -76,6 +76,7 @@ package object expressions {
     registry.registerExpression[GetExtent]("rf_extent")
     registry.registerExpression[GetCRS]("rf_crs")
     registry.registerExpression[RealizeTile]("rf_tile")
+    registry.registerExpression[CreateProjectedRaster]("rf_proj_raster")
     registry.registerExpression[Subtract]("rf_local_subtract")
     registry.registerExpression[Multiply]("rf_local_multiply")
     registry.registerExpression[Divide]("rf_local_divide")

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

@@ -0,0 +1,75 @@
+/*
+ * 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.transformers
+
+import geotrellis.proj4.CRS
+import geotrellis.vector.Extent
+import org.apache.spark.sql.catalyst.analysis.TypeCheckResult
+import org.apache.spark.sql.catalyst.analysis.TypeCheckResult.{TypeCheckFailure, TypeCheckSuccess}
+import org.apache.spark.sql.catalyst.expressions.codegen.CodegenFallback
+import org.apache.spark.sql.catalyst.expressions.{Expression, ExpressionDescription, TernaryExpression}
+import org.apache.spark.sql.types.DataType
+import org.apache.spark.sql.{Column, TypedColumn}
+import org.locationtech.rasterframes.encoders.CatalystSerializer._
+import org.locationtech.rasterframes.expressions.DynamicExtractors.tileExtractor
+import org.locationtech.rasterframes.expressions.row
+import org.locationtech.rasterframes.tiles.ProjectedRasterTile
+
+@ExpressionDescription(
+  usage = "_FUNC_(extent, crs, tile) - Construct a `proj_raster` structure from individual CRS, Extent, and Tile columns",
+  arguments = """
+  Arguments:
+    * extent - extent component of `proj_raster`
+    * crs - crs component of `proj_raster`
+    * tile - tile component of `proj_raster`"""
+)
+case class CreateProjectedRaster(tile: Expression, extent: Expression, crs: Expression) extends TernaryExpression with CodegenFallback {
+  override def nodeName: String = "rf_proj_raster"
+
+  override def children: Seq[Expression] = Seq(tile, extent, crs)
+
+  override def dataType: DataType = schemaOf[ProjectedRasterTile]
+
+  override def checkInputDataTypes(): TypeCheckResult = {
+    if (!tileExtractor.isDefinedAt(tile.dataType)) {
+      TypeCheckFailure(s"Column of type '${tile.dataType}' is not or does not have a Tile")
+    }
+    else if (!extent.dataType.conformsTo[Extent]) {
+      TypeCheckFailure(s"Column of type '${extent.dataType}' is not an Extent")
+    }
+    else if (!crs.dataType.conformsTo[CRS]) {
+      TypeCheckFailure(s"Column of type '${crs.dataType}' is not a CRS")
+    }
+    else TypeCheckSuccess
+  }
+  override protected def nullSafeEval(tileInput: Any, extentInput: Any, crsInput: Any): Any = {
+    val e = row(extentInput).to[Extent]
+    val c = row(crsInput).to[CRS]
+    val (t, _) = tileExtractor(tile.dataType)(row(tileInput))
+    ProjectedRasterTile(t, e, c).toInternalRow
+  }
+}
+
+object CreateProjectedRaster {
+  def apply(tile: Column, extent: Column, crs: Column): TypedColumn[Any, ProjectedRasterTile] =
+    new Column(new CreateProjectedRaster(tile.expr, extent.expr, crs.expr)).as[ProjectedRasterTile]
+}