Cleanup ScalaUDF usage

Author: pomadchin

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

@@ -22,8 +22,7 @@
 package org.apache.spark.sql.rf
 
 import org.apache.spark.sql.catalyst.InternalRow
-import org.apache.spark.sql.types.{DataType, UDTRegistration, UserDefinedType, _}
-import org.locationtech.rasterframes.expressions.transformers.RasterRefToTile
+import org.apache.spark.sql.types._
 import org.locationtech.rasterframes.ref.RFRasterSource
 import org.locationtech.rasterframes.util.KryoSupport
 

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

@@ -40,8 +40,8 @@ trait UnaryRasterAggregate extends DeclarativeAggregate {
 
   def children = Seq(child)
 
-  protected def tileOpAsExpressionNew[R: TypeTag](name: String, op: Tile => R): Expression => ScalaUDF =
-    udfexprNew[R, Any](name, (dataType: DataType) => (a: Any) => if(a == null) null.asInstanceOf[R] else op(UnaryRasterAggregate.extractTileFromAny(dataType, a)))
+  protected def tileOpAsExpression[R: TypeTag](name: String, op: Tile => R): Expression => ScalaUDF =
+    udfexpr[R, Any, Tile](name, (dataType: DataType) => (a: Any) => if(a == null) null.asInstanceOf[R] else op(UnaryRasterAggregate.extractTileFromAny(dataType, a)))
 }
 
 object UnaryRasterAggregate {

core/src/main/scala/org/locationtech/rasterframes/expressions/accessors/GetCRS.scala

@@ -21,7 +21,7 @@
 
 package org.locationtech.rasterframes.expressions.accessors
 
-import geotrellis.proj4.{CRS, LatLng}
+import geotrellis.proj4.CRS
 import org.apache.spark.sql.catalyst.InternalRow
 import org.apache.spark.sql.catalyst.analysis.TypeCheckResult
 import org.apache.spark.sql.catalyst.analysis.TypeCheckResult.{TypeCheckFailure, TypeCheckSuccess}

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

@@ -43,8 +43,8 @@ abstract class CellCountAggregate(isData: Boolean) extends UnaryRasterAggregate
   val initialValues = Seq(Literal(0L))
 
   private def CellTest: Expression => ScalaUDF =
-    if (isData) tileOpAsExpressionNew("rf_data_cells", DataCells.op)
-    else tileOpAsExpressionNew("rf_no_data_cells", NoDataCells.op)
+    if (isData) tileOpAsExpression("rf_data_cells", DataCells.op)
+    else tileOpAsExpression("rf_no_data_cells", NoDataCells.op)
 
   val updateExpressions = Seq(If(IsNull(child), count, Add(count, CellTest(child))))
 

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

@@ -54,8 +54,8 @@ case class CellMeanAggregate(child: Expression) extends UnaryRasterAggregate {
   // Cant' figure out why we can't just use the Expression directly
   // this is necessary to properly handle null rows. For example,
   // if we use `tilestats.Sum` directly, we get an NPE when the stage is executed.
-  private val DataCellCounts = tileOpAsExpressionNew("rf_data_cells", DataCells.op)
-  private val SumCells = tileOpAsExpressionNew("sum_cells", Sum.op)
+  private val DataCellCounts = tileOpAsExpression("rf_data_cells", DataCells.op)
+  private val SumCells = tileOpAsExpression("sum_cells", Sum.op)
 
   val updateExpressions = Seq(
     // TODO: Figure out why this doesn't work. See above.

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

@@ -48,7 +48,7 @@ case class LocalMeanAggregate(child: Expression) extends UnaryRasterAggregate {
 
   def aggBufferAttributes: Seq[AttributeReference] = Seq(count, sum)
 
-  private lazy val Defined: Expression => ScalaUDF = tileOpAsExpressionNew("defined_cells", local.Defined.apply)
+  private lazy val Defined: Expression => ScalaUDF = tileOpAsExpression("defined_cells", local.Defined.apply)
 
   lazy val initialValues: Seq[Expression] = Seq(
     Literal.create(null, dataType),

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

@@ -53,21 +53,9 @@ package object expressions {
 
   /** As opposed to `udf`, this constructs an unwrapped ScalaUDF Expression from a function. */
   private[expressions]
-  def udfexpr[RT: TypeTag, A1: TypeTag](name: String, f: A1 => RT): Expression => ScalaUDF = (child: Expression) => {
+  def udfexpr[RT: TypeTag, A1: TypeTag, A1T: TypeTag](name: String, f: DataType => A1 => RT): Expression => ScalaUDF = (exp: Expression) => {
     val ScalaReflection.Schema(dataType, nullable) = ScalaReflection.schemaFor[RT]
-    ScalaUDF(f, dataType, Seq(child), Option(ExpressionEncoder[RT]()) :: Nil, udfName = Some(name))
-  }
-
-  private[expressions]
-  def udfexprNew[RT: TypeTag, A1: TypeTag](name: String, f: DataType => A1 => RT): Expression => ScalaUDF = (exp: Expression) => {
-    val ScalaReflection.Schema(dataType, nullable) = ScalaReflection.schemaFor[RT]
-    ScalaUDF((row: A1) => f(exp.dataType)(row), dataType, exp :: Nil, Option(ExpressionEncoder[RT]().resolveAndBind()) :: Nil)
-  }
-
-  private[expressions]
-  def udfexprNewUntyped[RT: TypeTag, A1: TypeTag](name: String, f: DataType => A1 => RT): Expression => ScalaUDF = (exp: Expression) => {
-    val ScalaReflection.Schema(dataType, nullable) = ScalaReflection.schemaFor[RT]
-    ScalaUDF((row: A1) => f(exp.dataType)(row), dataType, exp :: Nil)
+    ScalaUDF((row: A1) => f(exp.dataType)(row), dataType, exp :: Nil, Option(ExpressionEncoder[A1T]().resolveAndBind()) :: Nil)
   }
 
   def register(sqlContext: SQLContext): Unit = {

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

@@ -27,7 +27,6 @@ import geotrellis.vector.Extent
 import org.apache.spark.sql.functions.udf
 import org.apache.spark.sql.{Row, SQLContext}
 import org.locationtech.jts.geom.Geometry
-import org.locationtech.rasterframes._
 import org.locationtech.rasterframes.encoders.syntax._
 import org.locationtech.rasterframes.util.ResampleMethod
 

core/src/main/scala/org/locationtech/rasterframes/ref/RFRasterSource.scala

@@ -24,20 +24,17 @@ package org.locationtech.rasterframes.ref
 import java.net.URI
 import com.github.blemale.scaffeine.Scaffeine
 import com.typesafe.scalalogging.LazyLogging
-import frameless.Injection
 import geotrellis.proj4.CRS
 import geotrellis.raster._
 import geotrellis.raster.io.geotiff.Tags
 import geotrellis.vector.Extent
 import org.apache.hadoop.conf.Configuration
 import org.apache.spark.annotation.Experimental
 import org.apache.spark.sql.catalyst.encoders.ExpressionEncoder
-import org.apache.spark.sql.rf.{RasterSourceUDT}
+import org.apache.spark.sql.rf.RasterSourceUDT
 import org.locationtech.rasterframes.model.TileContext
-import org.locationtech.rasterframes.util.KryoSupport
 import org.locationtech.rasterframes.{NOMINAL_TILE_DIMS, rfConfig}
 
-import java.nio.ByteBuffer
 import scala.concurrent.duration.{Duration, FiniteDuration}
 
 /**