Merge branch 'develop' into fix/rf_ipy_render_383

Author: metasim

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

@@ -22,12 +22,11 @@ package org.locationtech.rasterframes.bench
 
 import java.util.concurrent.TimeUnit
 
-import org.apache.spark.sql.catalyst.InternalRow
-import org.apache.spark.sql.catalyst.expressions.BoundReference
-import org.apache.spark.sql.rf.TileUDT
 import org.locationtech.rasterframes._
-import org.locationtech.rasterframes.expressions.generators.ExplodeTiles
+import org.apache.spark.sql._
+import org.apache.spark.sql.functions._
 import org.openjdk.jmh.annotations._
+
 /**
  *
  * @author sfitch
@@ -37,32 +36,33 @@ import org.openjdk.jmh.annotations._
 @State(Scope.Benchmark)
 @OutputTimeUnit(TimeUnit.MILLISECONDS)
 class TileExplodeBench extends SparkEnv {
+  import spark.implicits._
 
-  //@Param(Array("uint8", "uint16ud255", "float32", "float64"))
-  @Param(Array("uint16ud255"))
+  @Param(Array("uint8", "uint16ud255", "float32", "float64"))
   var cellTypeName: String = _
 
   @Param(Array("256"))
   var tileSize: Int = _
 
-  @Param(Array("2000"))
+  @Param(Array("100"))
   var numTiles: Int = _
 
   @transient
-  var tiles: Array[InternalRow] = _
-
-  var exploder: ExplodeTiles = _
+  var tiles: DataFrame = _
 
   @Setup(Level.Trial)
   def setupData(): Unit = {
-    tiles = Array.fill(numTiles)(randomTile(tileSize, tileSize, cellTypeName))
-        .map(t => InternalRow(TileUDT.tileSerializer.toInternalRow(t)))
-    val expr = BoundReference(0, TileType, true)
-    exploder = new ExplodeTiles(1.0, None, Seq(expr))
+    tiles = Seq.fill(numTiles)(randomTile(tileSize, tileSize, cellTypeName))
+      .toDF("tile").repartition(10)
+  }
+
+  @Benchmark
+  def arrayExplode() = {
+    tiles.select(posexplode(rf_tile_to_array_double($"tile"))).count()
   }
+
   @Benchmark
   def tileExplode() = {
-    for(t <- tiles)
-      exploder.eval(t)
+    tiles.select(rf_explode_tiles($"tile")).count()
   }
 }

build.sbt

@@ -32,7 +32,10 @@ lazy val root = project
   .withId("RasterFrames")
   .aggregate(core, datasource, pyrasterframes, experimental)
   .enablePlugins(RFReleasePlugin)
-  .settings(publish / skip := true)
+  .settings(
+    publish / skip := true,
+    clean := clean.dependsOn(`rf-notebook`/clean).value
+  )
 
 lazy val `rf-notebook` = project
   .dependsOn(pyrasterframes)

build/circleci/Dockerfile

@@ -14,16 +14,18 @@ RUN sudo apt-get update && \
       pandoc \
       wget \
       gcc g++ build-essential \
-      libreadline-gplv2-dev libncursesw5-dev libssl-dev libsqlite3-dev tk-dev libgdbm-dev libc6-dev libbz2-dev \
-            libcurl4-gnutls-dev \
-            libproj-dev \
-            libgeos-dev \
-            libhdf4-alt-dev \
-            bash-completion \
-            cmake \
-            imagemagick \
-            libpng-dev \
-            libffi-dev \
+      libreadline-gplv2-dev libncursesw5-dev \
+      libssl-dev libsqlite3-dev tk-dev libgdbm-dev libc6-dev libbz2-dev \
+      liblzma-dev \
+      libcurl4-gnutls-dev \
+      libproj-dev \
+      libgeos-dev \
+      libhdf4-alt-dev \
+      bash-completion \
+      cmake \
+      imagemagick \
+      libpng-dev \
+      libffi-dev \
     && sudo apt autoremove \
     && sudo apt-get clean all
 #    && sudo update-alternatives --install /usr/bin/python python /usr/bin/python3 1
@@ -73,8 +75,7 @@ RUN cd /tmp && \
         --with-threads \
         --without-jp2mrsid \
         --without-netcdf \
-        --without-ecw \
-    && \
+        --without-ecw && \
     make -j 8 && \
     sudo make install && \
     sudo ldconfig && \

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

@@ -123,8 +123,8 @@ object CellStatsAggregate {
   import org.locationtech.rasterframes.encoders.StandardEncoders.cellStatsEncoder
 
   def apply(col: Column): TypedColumn[Any, CellStatistics] =
-    new Column(new CellStatsAggregateUDAF(col.expr))
-      .as(s"rf_agg_stats($col)") // node renaming in class doesn't seem to propogate
+    new CellStatsAggregate()(ExtractTile(col))
+      .as(s"rf_agg_stats($col)")
       .as[CellStatistics]
 
   /** Adapter hack to allow UserDefinedAggregateFunction to be referenced as an expression. */

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

@@ -98,8 +98,8 @@ object HistogramAggregate {
   import org.locationtech.rasterframes.encoders.StandardEncoders.cellHistEncoder
 
   def apply(col: Column): TypedColumn[Any, CellHistogram] =
-    new Column(new HistogramAggregateUDAF(col.expr))
-      .as(s"rf_agg_approx_histogram($col)") // node renaming in class doesn't seem to propogate
+    new HistogramAggregate()(ExtractTile(col))
+      .as(s"rf_agg_approx_histogram($col)")
       .as[CellHistogram]
 
   /** Adapter hack to allow UserDefinedAggregateFunction to be referenced as an expression. */

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

@@ -92,7 +92,7 @@ object LocalCountAggregate {
   object LocalDataCellsUDAF {
     def apply(child: Expression): LocalDataCellsUDAF = new LocalDataCellsUDAF(child)
     def apply(tile: Column): TypedColumn[Any, Tile] =
-      new Column(new LocalDataCellsUDAF(tile.expr))
+      new LocalCountAggregate(true)(ExtractTile(tile))
         .as(s"rf_agg_local_data_cells($tile)")
         .as[Tile]
   }
@@ -107,7 +107,7 @@ object LocalCountAggregate {
   object LocalNoDataCellsUDAF {
     def apply(child: Expression): LocalNoDataCellsUDAF = new LocalNoDataCellsUDAF(child)
     def apply(tile: Column): TypedColumn[Any, Tile] =
-      new Column(new LocalNoDataCellsUDAF(tile.expr))
+      new LocalCountAggregate(false)(ExtractTile(tile))
         .as(s"rf_agg_local_no_data_cells($tile)")
         .as[Tile]
   }

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

@@ -146,7 +146,7 @@ class LocalStatsAggregate() extends UserDefinedAggregateFunction {
 object LocalStatsAggregate {
 
   def apply(col: Column): TypedColumn[Any, LocalCellStatistics] =
-    new Column(LocalStatsAggregateUDAF(col.expr))
+    new LocalStatsAggregate()(ExtractTile(col))
       .as(s"rf_agg_local_stats($col)")
       .as[LocalCellStatistics]
 

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

@@ -83,7 +83,10 @@ object LocalTileOpAggregate {
   }
   object LocalMinUDAF {
     def apply(child: Expression): LocalMinUDAF = new LocalMinUDAF(child)
-    def apply(tile: Column): TypedColumn[Any, Tile] = new Column(new LocalMinUDAF(tile.expr)).as[Tile]
+    def apply(tile: Column): TypedColumn[Any, Tile] =
+      new LocalTileOpAggregate(BiasedMin)(ExtractTile(tile))
+        .as(s"rf_agg_local_min($tile)")
+        .as[Tile]
   }
 
   @ExpressionDescription(
@@ -95,6 +98,9 @@ object LocalTileOpAggregate {
   }
   object LocalMaxUDAF {
     def apply(child: Expression): LocalMaxUDAF = new LocalMaxUDAF(child)
-    def apply(tile: Column): TypedColumn[Any, Tile] = new Column(new LocalMaxUDAF(tile.expr)).as[Tile]
+    def apply(tile: Column): TypedColumn[Any, Tile] =
+      new LocalTileOpAggregate(BiasedMax)(ExtractTile(tile))
+        .as(s"rf_agg_local_max($tile)")
+        .as[Tile]
   }
 }

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

@@ -24,8 +24,8 @@ package org.locationtech.rasterframes.expressions.generators
 import geotrellis.raster._
 import org.apache.spark.sql._
 import org.apache.spark.sql.catalyst.InternalRow
-import org.apache.spark.sql.catalyst.expressions.codegen.{BufferHolder, CodegenFallback, UnsafeRowWriter}
-import org.apache.spark.sql.catalyst.expressions.{Expression, Generator, UnsafeRow}
+import org.apache.spark.sql.catalyst.expressions.codegen.CodegenFallback
+import org.apache.spark.sql.catalyst.expressions.{Expression, Generator, GenericInternalRow}
 import org.apache.spark.sql.types._
 import org.locationtech.rasterframes._
 import org.locationtech.rasterframes.expressions.DynamicExtractors
@@ -87,17 +87,14 @@ case class ExplodeTiles(
       cfor(0)(_ < rows, _ + 1) { row =>
         cfor(0)(_ < cols, _ + 1) { col =>
           val rowIndex = row * cols + col
-          val outRow = new UnsafeRow(numOutCols)
-          val buffer = new BufferHolder(outRow)
-          val writer = new UnsafeRowWriter(buffer, numOutCols)
-          writer.write(0, col)
-          writer.write(1, row)
+          val outCols = Array.ofDim[Any](numOutCols)
+          outCols(0) = col
+          outCols(1) = row
           cfor(0)(_ < tiles.length, _ + 1) { index =>
             val tile = tiles(index)
-            val cell: Double = if (tile == null) doubleNODATA else tile.getDouble(col, row)
-            writer.write(index + 2, cell)
+            outCols(index + 2) = if(tile == null) doubleNODATA else tile.getDouble(col, row)
           }
-          retval(rowIndex) = outRow
+          retval(rowIndex) = new GenericInternalRow(outCols)
         }
       }
       if(sampleFraction > 0.0 && sampleFraction < 1.0) sample(retval)

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

@@ -36,7 +36,7 @@ import org.locationtech.rasterframes.expressions.tilestats._
 import org.locationtech.rasterframes.expressions.transformers._
 
 import scala.reflect.runtime.universe._
-import scala.util.Try
+
 /**
  * Module of Catalyst expressions for efficiently working with tiles.
  *
@@ -53,8 +53,7 @@ package object expressions {
   private[expressions]
   def udfexpr[RT: TypeTag, A1: TypeTag](name: String, f: A1 => RT): Expression => ScalaUDF = (child: Expression) => {
     val ScalaReflection.Schema(dataType, nullable) = ScalaReflection.schemaFor[RT]
-    val inputTypes = Try(ScalaReflection.schemaFor(typeTag[A1]).dataType :: Nil).toOption
-    ScalaUDF(f, dataType, Seq(child),  inputTypes.getOrElse(Nil), nullable = nullable, udfName = Some(name))
+    ScalaUDF(f, dataType, Seq(child), Seq(true), nullable = nullable, udfName = Some(name))
   }
 
   def register(sqlContext: SQLContext): Unit = {