Cleanup datasource project

Author: pomadchin

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

@@ -26,7 +26,6 @@ import java.util.concurrent.TimeUnit
 import geotrellis.raster.Dimensions
 import org.apache.spark.sql.catalyst.InternalRow
 import org.apache.spark.sql.rf.TileUDT
-import org.locationtech.rasterframes.tiles.InternalRowTile
 import org.openjdk.jmh.annotations._
 
 @BenchmarkMode(Array(Mode.AverageTime))
@@ -62,15 +61,4 @@ class TileCellScanBench extends SparkEnv {
       tile.getDouble(cols/2, rows/2) +
       tile.getDouble(0, 0)
   }
-
-  @Benchmark
-  def internalRowRead(): Double  = {
-    val tile = new InternalRowTile(tileRow)
-    val cols = tile.cols
-    val rows = tile.rows
-    tile.getDouble(cols - 1, rows - 1) +
-      tile.getDouble(cols/2, rows/2) +
-      tile.getDouble(0, 0)
-  }
 }
-

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

@@ -59,4 +59,4 @@ object ProjectedRasterTile {
     Some((prt.tile, prt.extent, prt.crs))
 
   implicit lazy val projectedRasterTileEncoder: ExpressionEncoder[ProjectedRasterTile] = ExpressionEncoder[ProjectedRasterTile]()
-}
+}

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

@@ -23,7 +23,6 @@ package org.locationtech.rasterframes
 
 import geotrellis.raster
 import geotrellis.raster.{CellType, Dimensions, NoNoData, Tile}
-import org.apache.spark.sql.catalyst.encoders.ExpressionEncoder
 import org.apache.spark.sql.types.StringType
 import org.locationtech.rasterframes.tiles.ShowableTile
 import org.scalatest.Inspectors
@@ -37,7 +36,6 @@ class TileUDTSpec extends TestEnvironment with TestData with Inspectors {
   import TestData.randomTile
 
   spark.version
-  val tileEncoder: ExpressionEncoder[Tile] = ExpressionEncoder()
 
   describe("TileUDT") {
     val tileSizes = Seq(2, 7, 64, 128, 511)
@@ -97,7 +95,7 @@ class TileUDTSpec extends TestEnvironment with TestData with Inspectors {
 
       if (rfConfig.getBoolean("showable-tiles"))
         forEveryConfig { tile =>
-          val stringified = Seq(Option(tile)).toDF("tile").select($"tile".cast(StringType)).as[String].first()
+          val stringified = Seq(tile).toDF("tile").select($"tile".cast(StringType)).as[String].first()
           stringified should be(ShowableTile.show(tile))
 
           if(!tile.cellType.isInstanceOf[NoNoData]) {

datasource/src/main/scala/org/locationtech/rasterframes/datasource/geotiff/GeoTiffRelation.scala

@@ -21,25 +21,26 @@
 
 package org.locationtech.rasterframes.datasource.geotiff
 
-import java.net.URI
-import com.typesafe.scalalogging.Logger
 import geotrellis.layer._
 import geotrellis.spark._
 import geotrellis.store.hadoop.util.HdfsRangeReader
 import org.apache.hadoop.fs.Path
 import org.apache.spark.rdd.RDD
-import org.apache.spark.sql.catalyst.encoders.{ExpressionEncoder, RowEncoder}
+import org.apache.spark.sql.catalyst.encoders.ExpressionEncoder
 import org.apache.spark.sql.rf.TileUDT
 import org.apache.spark.sql.sources._
 import org.apache.spark.sql.types._
 import org.apache.spark.sql.{Row, SQLContext}
-import org.locationtech.rasterframes._
 import org.locationtech.rasterframes.util._
 import org.slf4j.LoggerFactory
 import JsonCodecs._
 import geotrellis.raster.CellGrid
 import geotrellis.spark.store.hadoop.{HadoopGeoTiffRDD, HadoopGeoTiffReader}
-import org.locationtech.rasterframes.encoders.StandardEncoders
+import org.locationtech.rasterframes._
+import org.locationtech.rasterframes.encoders.syntax._
+
+import java.net.URI
+import com.typesafe.scalalogging.Logger
 
 /**
  * Spark SQL data source over a single GeoTiff file. Works best with CoG compliant ones.
@@ -71,11 +72,9 @@ case class GeoTiffRelation(sqlContext: SQLContext, uri: URI) extends BaseRelatio
 
     StructType(Seq(
       StructField(SPATIAL_KEY_COLUMN.columnName, skSchema, nullable = false, skMetadata),
-      StructField(EXTENT_COLUMN.columnName, StandardEncoders.extentEncoder.schema, nullable = true),
-      StructField(CRS_COLUMN.columnName, CrsType, nullable = true),
-      StructField(METADATA_COLUMN.columnName,
-        DataTypes.createMapType(StringType, StringType, false)
-      )
+      StructField(EXTENT_COLUMN.columnName, extentEncoder.schema, nullable = true),
+      StructField(CRS_COLUMN.columnName, crsUDT, nullable = true),
+      StructField(METADATA_COLUMN.columnName, DataTypes.createMapType(StringType, StringType, false))
     ) ++ tileCols)
   }
 
@@ -90,14 +89,7 @@ case class GeoTiffRelation(sqlContext: SQLContext, uri: URI) extends BaseRelatio
     val trans = tlm.mapTransform
     val metadata = info.tags.headTags
 
-    val encodedCRS =
-      RowEncoder(StandardEncoders.crsSparkEncoder.schema)
-        .resolveAndBind()
-        .createDeserializer()(
-          StandardEncoders
-            .crsSparkEncoder
-            .createSerializer()(tlm.crs)
-        )
+    val encodedCRS = tlm.crs.toRow
 
     if(info.segmentLayout.isTiled) {
       // TODO: Figure out how to do tile filtering via the range reader.
@@ -108,22 +100,8 @@ case class GeoTiffRelation(sqlContext: SQLContext, uri: URI) extends BaseRelatio
           // transform result because the layout is directly from the TIFF
           val gb = trans.extentToBounds(pe.extent)
           val entries = columnIndexes.map {
-            case 0 =>
-              RowEncoder(StandardEncoders.spatialKeyEncoder.schema)
-                .resolveAndBind()
-                .createDeserializer()(
-                  StandardEncoders
-                    .spatialKeyEncoder
-                    .createSerializer()(SpatialKey(gb.colMin, gb.rowMin))
-                )
-            case 1 =>
-              RowEncoder(StandardEncoders.extentEncoder.schema)
-                .resolveAndBind()
-                .createDeserializer()(
-                  StandardEncoders
-                    .extentEncoder
-                    .createSerializer()(pe.extent)
-                )
+            case 0 => SpatialKey(gb.colMin, gb.rowMin).toRow
+            case 1 => pe.extent.toRow
             case 2 => encodedCRS
             case 3 => metadata
             case n => tiles.band(n - 4)
@@ -147,14 +125,7 @@ case class GeoTiffRelation(sqlContext: SQLContext, uri: URI) extends BaseRelatio
         .map { case (sk, tiles) =>
           val entries = columnIndexes.map {
             case 0 => sk
-            case 1 =>
-              RowEncoder(StandardEncoders.extentEncoder.schema)
-                .resolveAndBind()
-                .createDeserializer()(
-                  StandardEncoders
-                    .extentEncoder
-                    .createSerializer()(trans.keyToExtent(sk))
-                )
+            case 1 => trans.keyToExtent(sk).toRow
             case 2 => encodedCRS
             case 3 => metadata
             case n => tiles.band(n - 4)

datasource/src/main/scala/org/locationtech/rasterframes/datasource/geotrellis/Layer.scala

@@ -21,12 +21,13 @@
 
 package org.locationtech.rasterframes.datasource.geotrellis
 
+import org.locationtech.rasterframes._
+import org.apache.spark.sql.catalyst.encoders.ExpressionEncoder
+
+import geotrellis.store.LayerId
 import frameless.TypedEncoder
 
 import java.net.URI
-import org.locationtech.rasterframes.encoders.typedExpressionEncoder
-import geotrellis.store.LayerId
-import org.apache.spark.sql.catalyst.encoders.ExpressionEncoder
 
 /**
  *   /** Connector between a GT `LayerId` and the path in which it lives. */
@@ -36,11 +37,9 @@ import org.apache.spark.sql.catalyst.encoders.ExpressionEncoder
 case class Layer(base: URI, id: LayerId)
 
 object Layer {
-  import org.locationtech.rasterframes.encoders.StandardEncoders._
-
   def apply(base: URI, name: String, zoom: Int) = new Layer(base, LayerId(name, zoom))
 
-  implicit def typedLayerEncoder: TypedEncoder[Layer] = TypedEncoder.usingDerivation
+  implicit val typedLayerEncoder: TypedEncoder[Layer] = TypedEncoder.usingDerivation
 
-  implicit def layerEncoder: ExpressionEncoder[Layer] = typedExpressionEncoder[Layer]
+  implicit val layerEncoder: ExpressionEncoder[Layer] = typedExpressionEncoder[Layer]
 }

datasource/src/main/scala/org/locationtech/rasterframes/datasource/raster/RasterSourceRelation.scala

@@ -82,7 +82,7 @@ case class RasterSourceRelation(
     sqlContext.sparkSession.sessionState.conf.numShufflePartitions
 
   override def schema: StructType = {
-    val tileSchema = ProjectedRasterTile.prtEncoder.schema
+    val tileSchema = ProjectedRasterTile.projectedRasterTileEncoder.schema
     val paths = for {
       pathCol <- pathColNames
     } yield StructField(pathCol, StringType, false)

datasource/src/test/scala/examples/ExplodeWithLocation.scala

@@ -24,12 +24,10 @@ package examples
 import geotrellis.raster._
 import geotrellis.vector.Extent
 import org.apache.spark.sql._
-import org.apache.spark.sql.catalyst.encoders.RowEncoder
 import org.apache.spark.sql.functions._
 import org.locationtech.rasterframes._
+import org.locationtech.rasterframes.encoders.syntax._
 import org.locationtech.rasterframes.datasource.raster._
-import org.locationtech.rasterframes.encoders.CatalystSerializer._
-import org.locationtech.rasterframes.encoders.StandardEncoders
 
 object ExplodeWithLocation extends App {
 
@@ -44,20 +42,8 @@ object ExplodeWithLocation extends App {
   val rf = spark.read.raster.from(example).withTileDimensions(16, 16).load()
 
   val grid2map = udf((encExtent: Row, encDims: Row, colIdx: Int, rowIdx: Int) => {
-    val extent =
-      extentEncoder
-        .resolveAndBind()
-        .createDeserializer()(
-          RowEncoder(extentEncoder.schema)
-            .createSerializer()(encExtent)
-        )
-    val dims =
-      dimensionsEncoder
-        .resolveAndBind()
-        .createDeserializer()(
-          RowEncoder(dimensionsEncoder.schema)
-            .createSerializer()(encDims)
-        )
+    val extent = encExtent.as[Extent]
+    val dims = encDims.as[Dimensions[Int]]
     GridExtent(extent, dims.cols, dims.rows).gridToMap(colIdx, rowIdx)
   })
 

datasource/src/test/scala/examples/ValueAtPoint.scala

@@ -24,11 +24,10 @@ package examples
 import org.apache.spark.sql._
 import org.apache.spark.sql.functions._
 import org.locationtech.rasterframes._
+import org.locationtech.rasterframes.encoders.syntax._
 import org.locationtech.rasterframes.datasource.raster._
-import org.locationtech.rasterframes.encoders.CatalystSerializer._
 import geotrellis.raster._
 import geotrellis.vector.Extent
-import org.apache.spark.sql.catalyst.encoders.RowEncoder
 import org.locationtech.jts.geom.Point
 
 object ValueAtPoint extends App {
@@ -45,13 +44,7 @@ object ValueAtPoint extends App {
   val point = st_makePoint(766770.000, 3883995.000)
 
   val rf_value_at_point = udf((extentEnc: Row, tile: Tile, point: Point) => {
-    val extent =
-      extentEncoder
-        .resolveAndBind()
-        .createDeserializer()(
-          RowEncoder(extentEncoder.schema)
-            .createSerializer()(extentEnc)
-        )
+    val extent = extentEnc.as[Extent]
     Raster(tile, extent).getDoubleValueAtPoint(point)
   })