Use spark-testing-base - core tests green

- fixed weird init order in tests
- all tests share same context now thanks to base
- exclude scala-xml from tests

Author: echeipesh

build.sbt

@@ -78,7 +78,7 @@ lazy val core = project
         ExclusionRule(organization = "com.github.mpilquist")
       ),
       scaffeine,
-      scalatest,
+      sparktestingbase excludeAll ExclusionRule("org.scala-lang.modules", "scala-xml_2.12"),
       `scala-logging`
     ),
     libraryDependencies ++= {

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

@@ -137,7 +137,7 @@ object TileRasterizerAggregate {
 
     destExtent.map { ext => c.copy(destinationExtent = ext) }
 
-    val aggs = tileCols.map(t => TileRasterizerAggregate(config, rf_crs(crsCol), extCol, rf_tile(t)).as(t.columnName))
+    val aggs = tileCols.map(t => TileRasterizerAggregate(config, rf_tile(t), extCol, rf_crs(crsCol)).as(t.columnName))
 
     val agg = df.select(aggs: _*)
 

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

@@ -27,8 +27,6 @@ import org.scalatest.Inspectors
 
 class BaseUdtSpec extends TestEnvironment with TestData with Inspectors {
 
-  spark.version
-
   it("should (de)serialize CRS") {
     val udt = new CrsUDT()
     val in = geotrellis.proj4.LatLng
@@ -37,6 +35,5 @@ class BaseUdtSpec extends TestEnvironment with TestData with Inspectors {
     out shouldBe in
     assert(out.isInstanceOf[LazyCRS])
     info(out.toString())
-
   }
 }

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

@@ -28,7 +28,6 @@ import org.locationtech.rasterframes.ref.RFRasterSource
 import org.locationtech.rasterframes.ref.RasterRef
 
 class CrsSpec extends TestEnvironment with TestData with Inspectors {
-  spark.version
   import spark.implicits._
 
   describe("CrsUDT") {

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

@@ -32,10 +32,8 @@ import org.locationtech.jts.geom.{Coordinate, GeometryFactory}
  * @since 12/16/17
  */
 class GeometryFunctionsSpec extends TestEnvironment with TestData with StandardColumns {
-  import spark.implicits._
-
   describe("Vector geometry operations") {
-    val rf = l8Sample(1).projectedRaster.toLayer(10, 10).withGeometry()
+    lazy val rf = l8Sample(1).projectedRaster.toLayer(10, 10).withGeometry()
     it("should allow joining and filtering of tiles based on points") {
       import spark.implicits._
 
@@ -136,6 +134,7 @@ class GeometryFunctionsSpec extends TestEnvironment with TestData with StandardC
   }
 
   it("should rasterize geometry") {
+    import spark.implicits._
     val rf = l8Sample(1).projectedRaster.toLayer.withGeometry()
     val df = GeomData.features.map(f => (
       f.geom.reproject(LatLng, rf.crs),

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

@@ -22,11 +22,10 @@
 package org.locationtech.rasterframes
 
 import geotrellis.raster._
-import geotrellis.raster.testkit.RasterMatchers
 import org.apache.spark.sql.functions._
 import org.locationtech.rasterframes.tiles.ProjectedRasterTile
 
-class RasterFunctionsSpec extends TestEnvironment with RasterMatchers {
+class RasterFunctionsSpec extends TestEnvironment {
   import TestData._
   import spark.implicits._
 

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

@@ -23,7 +23,6 @@ package org.locationtech.rasterframes
 
 import geotrellis.proj4.CRS
 import geotrellis.raster.resample._
-import geotrellis.raster.testkit.RasterMatchers
 import geotrellis.raster.{Dimensions, IntConstantNoDataCellType, Raster, Tile}
 import geotrellis.vector.Extent
 import org.apache.spark.SparkConf
@@ -32,18 +31,18 @@ import org.locationtech.rasterframes.expressions.aggregates.TileRasterizerAggreg
 import org.locationtech.rasterframes.expressions.aggregates.TileRasterizerAggregate.ProjectedRasterDefinition
 
 
-class RasterJoinSpec extends TestEnvironment with TestData with RasterMatchers {
-  import spark.implicits._
+class RasterJoinSpec extends TestEnvironment with TestData {
   describe("Raster join between two DataFrames") {
     val b4nativeTif = readSingleband("L8-B4-Elkton-VA.tiff")
     // Same data, reprojected to EPSG:4326
     val b4warpedTif = readSingleband("L8-B4-Elkton-VA-4326.tiff")
 
-    val b4nativeRf = b4nativeTif.toDF(Dimensions(10, 10))
-    val b4warpedRf = b4warpedTif.toDF(Dimensions(10, 10))
+    lazy val b4nativeRf = b4nativeTif.toDF(Dimensions(10, 10))
+    lazy val b4warpedRf = b4warpedTif.toDF(Dimensions(10, 10))
       .withColumnRenamed("tile", "tile2")
 
     it("should join the same scene correctly") {
+      import spark.implicits._
       val b4nativeRfPrime = b4nativeTif.toDF(Dimensions(10, 10))
         .withColumnRenamed("tile", "tile2")
       val joined = b4nativeRf.rasterJoin(b4nativeRfPrime.hint("broadcast"))
@@ -59,6 +58,7 @@ class RasterJoinSpec extends TestEnvironment with TestData with RasterMatchers {
     }
 
     it("should join same scene in different tile sizes"){
+      import spark.implicits._
       val r1prime = b4nativeTif.toDF(Dimensions(25, 25)).withColumnRenamed("tile", "tile2")
       r1prime.select(rf_dimensions($"tile2").getField("rows")).as[Int].first() should be (25)
       val joined = b4nativeRf.rasterJoin(r1prime)
@@ -75,6 +75,7 @@ class RasterJoinSpec extends TestEnvironment with TestData with RasterMatchers {
     }
 
     it("should join same scene in two projections, same tile size") {
+      import spark.implicits._
       val srcExtent = b4nativeTif.extent
       // b4warpedRf source data is gdal warped b4nativeRf data; join them together.
       val joined = b4nativeRf.rasterJoin(b4warpedRf)
@@ -112,6 +113,7 @@ class RasterJoinSpec extends TestEnvironment with TestData with RasterMatchers {
      }
 
     it("should join multiple RHS tile columns"){
+      import spark.implicits._
       // join multiple native CRS bands to the EPSG 4326 RF
 
       val multibandRf = b4nativeRf
@@ -126,6 +128,7 @@ class RasterJoinSpec extends TestEnvironment with TestData with RasterMatchers {
     }
 
     it("should join with heterogeneous LHS CRS and coverages"){
+      import spark.implicits._
 
       val df17 = readSingleband("m_3607824_se_17_1_20160620_subset.tif")
         .toDF(Dimensions(50, 50))
@@ -165,6 +168,7 @@ class RasterJoinSpec extends TestEnvironment with TestData with RasterMatchers {
     }
 
     it("should handle proj_raster types") {
+      import spark.implicits._
       val df1 = Seq(Option(one)).toDF("one")
       val df2 = Seq(Option(two)).toDF("two")
       noException shouldBe thrownBy {
@@ -174,6 +178,7 @@ class RasterJoinSpec extends TestEnvironment with TestData with RasterMatchers {
     }
 
     it("should raster join multiple times on projected raster"){
+      import spark.implicits._
       val df0 = Seq(Option(one)).toDF("proj_raster")
       val result = df0.select($"proj_raster" as "t1")
         .rasterJoin(df0.select($"proj_raster" as "t2"))
@@ -184,6 +189,7 @@ class RasterJoinSpec extends TestEnvironment with TestData with RasterMatchers {
     }
 
     it("should honor resampling options") {
+      import spark.implicits._
       // test case. replicate existing test condition and check that resampling option results in different output
       val filterExpr = st_intersects(rf_geometry($"tile"), st_point(704940.0, 4251130.0))
       val result = b4nativeRf.rasterJoin(b4warpedRf.withColumnRenamed("tile2", "nearest"), NearestNeighbor)
@@ -200,6 +206,7 @@ class RasterJoinSpec extends TestEnvironment with TestData with RasterMatchers {
     // Failed to execute user defined function(package$$$Lambda$4417/0x00000008019e2840: (struct<xmax:double,xmin:double,ymax:double,ymin:double>, string, array<struct<cellType:string,cols:int,rows:int,cells:binary,ref:struct<source:struct<raster_source_kryo:binary>,bandIndex:int,subextent:struct<xmin:double,ymin:double,xmax:double,ymax:double>,subgrid:struct<colMin:int,rowMin:int,colMax:int,rowMax:int>>>>, array<struct<xmax:double,xmin:double,ymax:double,ymin:double>>, array<string>, struct<cols:int,rows:int>, string) => struct<cellType:string,cols:int,rows:int,cells:binary,ref:struct<source:struct<raster_source_kryo:binary>,bandIndex:int,subextent:struct<xmin:double,ymin:double,xmax:double,ymax:double>,subgrid:struct<colMin:int,rowMin:int,colMax:int,rowMax:int>>>)
 
     it("should raster join with null left head") {
+      import spark.implicits._
       // https://github.com/locationtech/rasterframes/issues/462
       val prt = TestData.projectedRasterTile(
         10, 10, 1,
@@ -264,5 +271,5 @@ class RasterJoinSpec extends TestEnvironment with TestData with RasterMatchers {
 
   }
 
-  override def additionalConf: SparkConf = super.additionalConf.set("spark.sql.codegen.comments", "true")
+  override def additionalConf(conf: SparkConf) = conf.set("spark.sql.codegen.comments", "true")
 }

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

@@ -31,14 +31,10 @@ import org.locationtech.geomesa.curve.Z2SFC
  * @since 12/15/17
  */
 class SpatialKeySpec extends TestEnvironment with TestData {
-  assert(!spark.sparkContext.isStopped)
-
-  import spark.implicits._
-
   describe("Spatial key conversions") {
     val raster = sampleGeoTiff.projectedRaster
     // Create a raster frame with a single row
-    val rf = raster.toLayer(raster.tile.cols, raster.tile.rows)
+    lazy val rf = raster.toLayer(raster.tile.cols, raster.tile.rows)
 
     it("should add an extent column") {
       val expected = raster.extent.toPolygon()
@@ -53,12 +49,14 @@ class SpatialKeySpec extends TestEnvironment with TestData {
     }
 
     it("should add a center lat/lng value") {
+      import spark.implicits._
       val expected = raster.extent.center.reproject(raster.crs, LatLng)
       val result = rf.withCenterLatLng().select($"center".as[(Double, Double)]).first
       assert( Point(result._1, result._2) === expected)
     }
 
     it("should add a z-index value") {
+      import spark.implicits._
       val center = raster.extent.center.reproject(raster.crs, LatLng)
       val expected = Z2SFC.index(center.x, center.y)
       val result = rf.withSpatialIndex().select($"spatial_index".as[Long]).first

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

@@ -37,7 +37,6 @@ import org.scalatest.Inspectors
 class StandardEncodersSpec extends TestEnvironment with TestData with Inspectors {
 
   it("Dimensions encoder") {
-    spark.version
     import spark.implicits._
     val data = Dimensions[Int](256, 256)
     val df = List(data).toDF()
@@ -47,7 +46,6 @@ class StandardEncodersSpec extends TestEnvironment with TestData with Inspectors
   }
 
   it("TileDataContext encoder") {
-    spark.version
     import spark.implicits._
     val data = TileDataContext(IntCellType, Dimensions[Int](256, 256))
     val df = List(data).toDF()
@@ -57,7 +55,6 @@ class StandardEncodersSpec extends TestEnvironment with TestData with Inspectors
   }
 
   it("ProjectedExtent encoder") {
-    spark.version
     import spark.implicits._
     val data = ProjectedExtent(Extent(0, 0, 1, 1), LatLng)
     val df = List(data).toDF()
@@ -68,7 +65,6 @@ class StandardEncodersSpec extends TestEnvironment with TestData with Inspectors
   }
 
   it("TileLayerMetadata encoder"){
-    spark.version
     import spark.implicits._
     val data = TileLayerMetadata(
       IntCellType,

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

@@ -20,8 +20,9 @@
  */
 package org.locationtech.rasterframes
 
-import java.nio.file.{Files, Path}
+import com.holdenkarau.spark.testing.DataFrameSuiteBase
 
+import java.nio.file.{Files, Path}
 import com.typesafe.scalalogging.Logger
 import geotrellis.raster.Tile
 import geotrellis.raster.render.{ColorMap, ColorRamps}
@@ -39,11 +40,10 @@ import org.scalactic.Tolerance
 import org.scalatest._
 import org.scalatest.funspec.AnyFunSpec
 import org.scalatest.matchers.should.Matchers
-
 import org.scalatest.matchers.{MatchResult, Matcher}
 import org.slf4j.LoggerFactory
 
-trait TestEnvironment extends AnyFunSpec with Matchers with Inspectors with Tolerance with RasterMatchers {
+trait TestEnvironment extends AnyFunSpec with DataFrameSuiteBase with Matchers with RasterMatchers with Inspectors with Tolerance {
   @transient protected lazy val logger = Logger(LoggerFactory.getLogger(getClass.getName))
 
 
@@ -56,22 +56,29 @@ trait TestEnvironment extends AnyFunSpec with Matchers with Inspectors with Tole
   // allow 2 retries, should stabilize CI builds. https://spark.apache.org/docs/2.4.7/submitting-applications.html#master-urls
   def sparkMaster: String = "local[*, 2]"
 
-  def additionalConf: SparkConf =
-    new SparkConf(false)
-      .set("spark.driver.port", "0")
-      .set("spark.hostPort", "0")
-      .set("spark.ui.enabled", "false")
-
-  implicit val spark: SparkSession =
-      SparkSession
-        .builder
-        .master(sparkMaster)
-        .withKryoSerialization
-        .config(additionalConf)
-        .getOrCreate()
-        .withRasterFrames
-
-  implicit def sc: SparkContext = spark.sparkContext
+  protected def additionalConf(conf: SparkConf): SparkConf = conf
+
+  override def conf: SparkConf = {
+    val base = new SparkConf().
+      setAppName("RasterFrames Test").
+      setMaster(sparkMaster).
+      set("spark.serializer", "org.apache.spark.serializer.KryoSerializer").
+      set("spark.kryo.registrator", "org.locationtech.rasterframes.util.RFKryoRegistrator").
+      set("spark.ui.enabled", "false").
+      set("spark.driver.port", "0").
+      set("spark.hostPort", "0").
+      set("spark.ui.enabled", "true")
+    additionalConf(base)
+  }
+
+  override def setup(sc: SparkContext): Unit = {
+    sc.setCheckpointDir(com.holdenkarau.spark.testing.Utils.createTempDir().toPath().toString)
+    sc.setLogLevel("ERROR")
+    org.locationtech.rasterframes.initRF(sqlContext)
+  }
+
+  implicit def sparkSession: SparkSession = spark
+  implicit def sparkContext: SparkContext = spark.sparkContext
 
   lazy val sql: String => DataFrame = spark.sql