Fix DataBricks serialization issues and code cleanup

Author: pomadchin

build.sbt

@@ -7,8 +7,9 @@ val sparkVersion      = "3.1.3"
 val catsVersion       = "2.6.1"
 val shapelessVersion  = "2.3.3"                                   // to be compatible with Spark 3.1.x
 val scalaTestVersion  = "3.2.11"
+val framelessVersion  = "0.11.1"
 val geomesaVersion    = "3.3.0"
-val geotrellisVersion = "3.6.1+0-6b5868af+20220321-1909-SNAPSHOT" //"3.6.1"
+val geotrellisVersion = "3.6.1+1-e69dfae5+20220322-1723-SNAPSHOT" //"3.6.1"
 
 lazy val commonSettings = Seq(
   scalaVersion       := scalaVersions.head,

core/src/main/scala/com/azavea/hiveless/serializers/HSerializer.scala

@@ -52,8 +52,22 @@ object HSerializer extends Serializable {
     def serialize: T => Any = s
   }
 
-  /** Derive HSerializer from ExpressionEncoder. */
-  implicit def expressionEncoderSerializer[T: TypeTag](implicit enc: ExpressionEncoder[T]): HSerializer[T] = new HSerializer[T] {
+  // format: off
+  /**
+   * Derive HSerializer from ExpressionEncoder.
+   * Intentionally not used for instances implementation, causes the following failure on DataBricks:
+   * org.apache.spark.SparkException: Job aborted due to stage failure: Task serialization failed: com.esotericsoftware.kryo.KryoException: java.util.ConcurrentModificationException
+   * Serialization trace:
+   *   classes (sun.misc.Launcher$AppClassLoader)
+   *   classloader (java.security.ProtectionDomain)
+   *   context (java.security.AccessControlContext)
+   *   acc (com.databricks.backend.daemon.driver.ClassLoaders$LibraryClassLoader)
+   *   classLoader (scala.reflect.runtime.JavaMirrors$JavaMirror)
+   *   mirror (scala.reflect.api.TypeTags$TypeTagImpl)
+   *   tg$1 (com.azavea.hiveless.serializers.HSerializer$$anon$2)
+   */
+  // format: on
+  def expressionEncoderSerializer[T: TypeTag](implicit enc: ExpressionEncoder[T]): HSerializer[T] = new HSerializer[T] {
     def dataType: DataType  = enc.schema
     def serialize: T => Any = _.toInternalRow
   }

core/src/main/scala/com/azavea/hiveless/spark/encoders/syntax/package.scala

@@ -25,7 +25,7 @@ import scala.reflect.runtime.universe.TypeTag
 /**
  * Source: https://github.com/locationtech/rasterframes/blob/0.10.1/core/src/main/scala/org/locationtech/rasterframes/encoders/syntax/package.scala
  */
-package object syntax {
+package object syntax extends Serializable {
   implicit class CachedExpressionOps[T](val self: T) extends AnyVal {
     def toInternalRow(implicit tag: TypeTag[T], encoder: ExpressionEncoder[T]): InternalRow = {
       val toRow = SerializersCache.serializer[T]

project/plugins.sbt

@@ -2,4 +2,4 @@ addSbtPlugin("com.timushev.sbt"  % "sbt-updates"    % "0.6.1")
 addSbtPlugin("de.heikoseeberger" % "sbt-header"     % "5.6.5")
 addSbtPlugin("org.scalameta"     % "sbt-scalafmt"   % "2.4.3")
 addSbtPlugin("com.github.sbt"    % "sbt-ci-release" % "1.5.10")
-addSbtPlugin("com.eed3si9n"      % "sbt-assembly"   % "1.1.0")
+addSbtPlugin("com.eed3si9n"      % "sbt-assembly"   % "1.2.0")

spatial-index/sql/createUDFs.sql

@@ -1,5 +1,5 @@
-CREATE OR REPLACE FUNCTION st_crsFromText as 'com.azavea.hiveless.spatial.index.st_crsFromText';
-CREATE OR REPLACE FUNCTION st_extentFromGeom as 'com.azavea.hiveless.spatial.index.st_extentFromGeom';
-CREATE OR REPLACE FUNCTION st_geomReproject as 'com.azavea.hiveless.spatial.index.st_geomReproject';
-CREATE OR REPLACE FUNCTION st_partitionCentroid as 'com.azavea.hiveless.spatial.index.st_partitionCentroid';
-CREATE OR REPLACE FUNCTION st_z2LatLon as 'com.azavea.hiveless.spatial.index.st_z2LatLon';
+CREATE OR REPLACE FUNCTION st_crsFromText as 'com.azavea.hiveless.spatial.index.ST_CrsFromText';
+CREATE OR REPLACE FUNCTION st_extentFromGeom as 'com.azavea.hiveless.spatial.index.ST_ExtentFromGeom';
+CREATE OR REPLACE FUNCTION st_geomReproject as 'com.azavea.hiveless.spatial.index.ST_GeomReproject';
+CREATE OR REPLACE FUNCTION st_partitionCentroid as 'com.azavea.hiveless.spatial.index.ST_PartitionCentroid';
+CREATE OR REPLACE FUNCTION st_z2LatLon as 'com.azavea.hiveless.spatial.index.ST_Z2LatLon';

spatial-index/src/main/scala/com/azavea/hiveless/spark/geotrellis/encoders/StandardEncoders.scala

@@ -25,8 +25,8 @@ import scala.reflect.runtime.universe.TypeTag
 trait StandardEncoders extends Serializable {
   def expressionEncoder[T: TypeTag]: ExpressionEncoder[T] = ExpressionEncoder()
 
-  implicit lazy val extentEncoder: ExpressionEncoder[Extent]   = expressionEncoder
-  implicit lazy val z2IndexEncoder: ExpressionEncoder[Z2Index] = expressionEncoder
+  implicit val extentEncoder: ExpressionEncoder[Extent]   = expressionEncoder
+  implicit val z2IndexEncoder: ExpressionEncoder[Z2Index] = expressionEncoder
 }
 
 object StandardEncoders extends StandardEncoders

spatial-index/src/main/scala/com/azavea/hiveless/spatial/index/ST_Z2LatLon.scala

@@ -23,16 +23,15 @@ import geotrellis.store.index.zcurve.Z2
 import org.locationtech.jts.geom.Geometry
 
 class ST_Z2LatLon extends HUDF[Geometry, Z2Index] {
-  import ST_Z2LatLon._
-
   val name: String = "st_z2LatLon"
-  def function = { geom =>
-    val env = geom.getEnvelopeInternal
-    Z2Index(z2index(env.getMinX, env.getMinY), z2index(env.getMaxX, env.getMaxY))
-  }
+  def function     = ST_Z2LatLon.function
 }
 
 object ST_Z2LatLon {
+  def function(geom: Geometry): Z2Index = {
+    val env = geom.getEnvelopeInternal
+    Z2Index(z2index(env.getMinX, env.getMinY), z2index(env.getMaxX, env.getMaxY))
+  }
   def scaleLat(lat: Double): Int          = ((lat + 90) / 180 * (1 << 30)).toInt
   def scaleLong(lng: Double): Int         = ((lng + 180) / 360 * (1 << 30)).toInt
   def z2index(x: Double, y: Double): Long = Z2(scaleLong(x), scaleLat(y)).z

spatial-index/src/main/scala/com/azavea/hiveless/spatial/index/package.scala

@@ -18,9 +18,12 @@ package com.azavea.hiveless.spatial
 
 import com.azavea.hiveless.serializers.{HConverter, HSerializer, UnaryDeserializer}
 import com.azavea.hiveless.serializers.syntax._
+import com.azavea.hiveless.spark.encoders.syntax._
 import com.azavea.hiveless.spark.geotrellis.encoders.StandardEncoders
 import cats.Id
+import com.azavea.hiveless.spark.geotrellis.Z2Index
 import geotrellis.proj4.CRS
+import geotrellis.vector.Extent
 import org.apache.spark.sql.types.{DataType, StringType}
 
 package object index extends StandardEncoders {
@@ -35,4 +38,17 @@ package object index extends StandardEncoders {
     def dataType: DataType    = StringType
     def serialize: CRS => Any = crs => crs.toProj4String.serialize
   }
+
+  /**
+   * HSerializer.expressionEncoderSerializer causes serialization issues on DataBricks. TODO: investigate this issue.
+   */
+  implicit def extentSerializer: HSerializer[Extent] = new HSerializer[Extent] {
+    def dataType: DataType       = extentEncoder.schema
+    def serialize: Extent => Any = _.toInternalRow
+  }
+
+  implicit def z2IndexSerializer: HSerializer[Z2Index] = new HSerializer[Z2Index] {
+    def dataType: DataType        = z2IndexEncoder.schema
+    def serialize: Z2Index => Any = _.toInternalRow
+  }
 }