Fix for columnar functions which accept Tile or ProjectedRasterTile,

but claimed all return values were `Tile`. This was problematic whenever
the `Tile` as reified, and the serialization of `ProjectedRasterTile` confused
the `Tile` deserializer.

Fixes #295

Author: metasim

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

@@ -58,7 +58,7 @@ class BinaryTileOpBench extends SparkEnv {
 
   @Benchmark
   def viaExpression(): Array[Tile] = {
-    tiles.select(Add($"left", $"right")).collect()
+    tiles.select(Add($"left", $"right").as[Tile]).collect()
   }
 
   @Benchmark

core/src/main/scala/org/locationtech/rasterframes/RasterFunctions.scala

@@ -73,8 +73,8 @@ trait RasterFunctions {
 
   @Experimental
   /** Convert array in `arrayCol` into a Tile of dimensions `cols` and `rows`*/
-  def rf_array_to_tile(arrayCol: Column, cols: Int, rows: Int) = withAlias("rf_array_to_tile", arrayCol)(
-    udf[Tile, AnyRef](F.arrayToTile(cols, rows)).apply(arrayCol)
+  def rf_array_to_tile(arrayCol: Column, cols: Int, rows: Int): TypedColumn[Any, Tile] = withTypedAlias("rf_array_to_tile")(
+    udf[Tile, AnyRef](F.arrayToTile(cols, rows)).apply(arrayCol).as[Tile]
   )
 
   /** Create a Tile from a column of cell data with location indexes and preform cell conversion. */
@@ -89,12 +89,10 @@ trait RasterFunctions {
   def rf_cell_type(col: Column): TypedColumn[Any, CellType] = GetCellType(col)
 
   /** Change the Tile's cell type */
-  def rf_convert_cell_type(col: Column, cellType: CellType): TypedColumn[Any, Tile] =
-    SetCellType(col, cellType)
+  def rf_convert_cell_type(col: Column, cellType: CellType): Column = SetCellType(col, cellType)
 
   /** Change the Tile's cell type */
-  def rf_convert_cell_type(col: Column, cellTypeName: String): TypedColumn[Any, Tile] =
-    SetCellType(col, cellTypeName)
+  def rf_convert_cell_type(col: Column, cellTypeName: String): Column = SetCellType(col, cellTypeName)
 
   /** Resample tile to different size based on scalar factor or tile whose dimension to match. Scalar less
     * than one will downsample tile; greater than one will upsample. Uses nearest-neighbor. */
@@ -110,18 +108,20 @@ trait RasterFunctions {
   /** Extract the extent of a RasterSource or ProjectedRasterTile as a Geometry type. */
   def rf_geometry(raster: Column): TypedColumn[Any, Geometry] = GetGeometry(raster)
 
-  /** Assign a `NoData` value to the Tiles. */
-  def rf_with_no_data(col: Column, nodata: Double): TypedColumn[Any, Tile] = withTypedAlias("rf_with_no_data", col)(
-    udf[Tile, Tile](F.withNoData(nodata)).apply(col)
-  )
+  /** Assign a `NoData` value to the tile column. */
+  def rf_with_no_data(col: Column, nodata: Double): Column = SetNoDataValue(col, nodata)
+
+  /** Assign a `NoData` value to the tile column. */
+  def rf_with_no_data(col: Column, nodata: Int): Column = SetNoDataValue(col, nodata)
+
+  /** Assign a `NoData` value to the tile column. */
+  def rf_with_no_data(col: Column, nodata: Column): Column = SetNoDataValue(col, nodata)
 
   /**  Compute the full column aggregate floating point histogram. */
-  def rf_agg_approx_histogram(col: Column): TypedColumn[Any, CellHistogram] =
-    HistogramAggregate(col)
+  def rf_agg_approx_histogram(col: Column): TypedColumn[Any, CellHistogram] = HistogramAggregate(col)
 
   /** Compute the full column aggregate floating point statistics. */
-  def rf_agg_stats(col: Column): TypedColumn[Any, CellStatistics] =
-    CellStatsAggregate(col)
+  def rf_agg_stats(col: Column): TypedColumn[Any, CellStatistics] = CellStatsAggregate(col)
 
   /** Computes the column aggregate mean. */
   def rf_agg_mean(col: Column) = CellMeanAggregate(col)
@@ -194,28 +194,28 @@ trait RasterFunctions {
   def rf_agg_local_no_data_cells(col: Column): TypedColumn[Any, Tile] = LocalCountAggregate.LocalNoDataCellsUDAF(col)
 
   /** Cellwise addition between two Tiles or Tile and scalar column. */
-  def rf_local_add(left: Column, right: Column): TypedColumn[Any, Tile] = Add(left, right)
+  def rf_local_add(left: Column, right: Column): Column = Add(left, right)
 
   /** Cellwise addition of a scalar value to a tile. */
-  def rf_local_add[T: Numeric](tileCol: Column, value: T): TypedColumn[Any, Tile] = Add(tileCol, value)
+  def rf_local_add[T: Numeric](tileCol: Column, value: T): Column = Add(tileCol, value)
 
   /** Cellwise subtraction between two Tiles. */
-  def rf_local_subtract(left: Column, right: Column): TypedColumn[Any, Tile] = Subtract(left, right)
+  def rf_local_subtract(left: Column, right: Column): Column = Subtract(left, right)
 
   /** Cellwise subtraction of a scalar value from a tile. */
-  def rf_local_subtract[T: Numeric](tileCol: Column, value: T): TypedColumn[Any, Tile]  = Subtract(tileCol, value)
+  def rf_local_subtract[T: Numeric](tileCol: Column, value: T): Column  = Subtract(tileCol, value)
 
   /** Cellwise multiplication between two Tiles. */
-  def rf_local_multiply(left: Column, right: Column): TypedColumn[Any, Tile] = Multiply(left, right)
+  def rf_local_multiply(left: Column, right: Column): Column = Multiply(left, right)
 
   /** Cellwise multiplication of a tile by a scalar value. */
-  def rf_local_multiply[T: Numeric](tileCol: Column, value: T): TypedColumn[Any, Tile] = Multiply(tileCol, value)
+  def rf_local_multiply[T: Numeric](tileCol: Column, value: T): Column = Multiply(tileCol, value)
 
   /** Cellwise division between two Tiles. */
-  def rf_local_divide(left: Column, right: Column): TypedColumn[Any, Tile] = Divide(left, right)
+  def rf_local_divide(left: Column, right: Column): Column = Divide(left, right)
 
   /** Cellwise division of a tile by a scalar value. */
-  def rf_local_divide[T: Numeric](tileCol: Column, value: T): TypedColumn[Any, Tile] = Divide(tileCol, value)
+  def rf_local_divide[T: Numeric](tileCol: Column, value: T): Column = Divide(tileCol, value)
 
   /** Perform an arbitrary GeoTrellis `LocalTileBinaryOp` between two Tile columns. */
   def rf_local_algebra(op: LocalTileBinaryOp, left: Column, right: Column): TypedColumn[Any, Tile] =
@@ -326,96 +326,72 @@ trait RasterFunctions {
     DebugRender.RenderMatrix(col)
 
   /** Cellwise less than value comparison between two tiles. */
-  def rf_local_less(left: Column, right: Column): TypedColumn[Any, Tile] =
-    Less(left, right)
+  def rf_local_less(left: Column, right: Column): Column = Less(left, right)
 
   /** Cellwise less than value comparison between a tile and a scalar. */
-  def rf_local_less[T: Numeric](tileCol: Column, value: T): TypedColumn[Any, Tile] =
-    Less(tileCol, value)
+  def rf_local_less[T: Numeric](tileCol: Column, value: T): Column = Less(tileCol, value)
 
   /** Cellwise less than or equal to value comparison between a tile and a scalar. */
-  def rf_local_less_equal(left: Column, right: Column): TypedColumn[Any, Tile]  =
-    LessEqual(left, right)
+  def rf_local_less_equal(left: Column, right: Column): Column = LessEqual(left, right)
 
   /** Cellwise less than or equal to value comparison between a tile and a scalar. */
-  def rf_local_less_equal[T: Numeric](tileCol: Column, value: T): TypedColumn[Any, Tile] =
-    LessEqual(tileCol, value)
+  def rf_local_less_equal[T: Numeric](tileCol: Column, value: T): Column = LessEqual(tileCol, value)
 
   /** Cellwise greater than value comparison between two tiles. */
-  def rf_local_greater(left: Column, right: Column): TypedColumn[Any, Tile] =
-    Greater(left, right)
+  def rf_local_greater(left: Column, right: Column): Column = Greater(left, right)
 
   /** Cellwise greater than value comparison between a tile and a scalar. */
-  def rf_local_greater[T: Numeric](tileCol: Column, value: T): TypedColumn[Any, Tile] =
-    Greater(tileCol, value)
-
+  def rf_local_greater[T: Numeric](tileCol: Column, value: T): Column = Greater(tileCol, value)
   /** Cellwise greater than or equal to value comparison between two tiles. */
-  def rf_local_greater_equal(left: Column, right: Column): TypedColumn[Any, Tile]  =
-    GreaterEqual(left, right)
+  def rf_local_greater_equal(left: Column, right: Column): Column  = GreaterEqual(left, right)
 
   /** Cellwise greater than or equal to value comparison between a tile and a scalar. */
-  def rf_local_greater_equal[T: Numeric](tileCol: Column, value: T): TypedColumn[Any, Tile] =
-    GreaterEqual(tileCol, value)
+  def rf_local_greater_equal[T: Numeric](tileCol: Column, value: T): Column = GreaterEqual(tileCol, value)
 
   /** Cellwise equal to value comparison between two tiles. */
-  def rf_local_equal(left: Column, right: Column): TypedColumn[Any, Tile]  =
-    Equal(left, right)
+  def rf_local_equal(left: Column, right: Column): Column = Equal(left, right)
 
   /** Cellwise equal to value comparison between a tile and a scalar. */
-  def rf_local_equal[T: Numeric](tileCol: Column, value: T): TypedColumn[Any, Tile] =
-    Equal(tileCol, value)
+  def rf_local_equal[T: Numeric](tileCol: Column, value: T): Column = Equal(tileCol, value)
 
   /** Cellwise inequality comparison between two tiles. */
-  def rf_local_unequal(left: Column, right: Column): TypedColumn[Any, Tile]  =
-    Unequal(left, right)
+  def rf_local_unequal(left: Column, right: Column): Column  = Unequal(left, right)
 
   /** Cellwise inequality comparison between a tile and a scalar. */
-  def rf_local_unequal[T: Numeric](tileCol: Column, value: T): TypedColumn[Any, Tile] =
-    Unequal(tileCol, value)
+  def rf_local_unequal[T: Numeric](tileCol: Column, value: T): Column = Unequal(tileCol, value)
 
   /** Round cell values to nearest integer without chaning cell type. */
-  def rf_round(tileCol: Column): TypedColumn[Any, Tile] =
-    Round(tileCol)
+  def rf_round(tileCol: Column): Column = Round(tileCol)
 
   /** Compute the absolute value of each cell. */
-  def rf_abs(tileCol: Column): TypedColumn[Any, Tile] =
-    Abs(tileCol)
+  def rf_abs(tileCol: Column): Column = Abs(tileCol)
 
   /** Take natural logarithm of cell values. */
-  def rf_log(tileCol: Column): TypedColumn[Any, Tile] =
-    Log(tileCol)
+  def rf_log(tileCol: Column): Column = Log(tileCol)
 
   /** Take base 10 logarithm of cell values. */
-  def rf_log10(tileCol: Column): TypedColumn[Any, Tile] =
-    Log10(tileCol)
+  def rf_log10(tileCol: Column): Column = Log10(tileCol)
 
   /** Take base 2 logarithm of cell values. */
-  def rf_log2(tileCol: Column): TypedColumn[Any, Tile] =
-    Log2(tileCol)
+  def rf_log2(tileCol: Column): Column = Log2(tileCol)
 
   /** Natural logarithm of one plus cell values. */
-  def rf_log1p(tileCol: Column): TypedColumn[Any, Tile] =
-      Log1p(tileCol)
+  def rf_log1p(tileCol: Column): Column = Log1p(tileCol)
 
   /** Exponential of cell values */
-  def rf_exp(tileCol: Column): TypedColumn[Any, Tile] =
-    Exp(tileCol)
+  def rf_exp(tileCol: Column): Column = Exp(tileCol)
 
   /** Ten to the power of cell values */
-  def rf_exp10(tileCol: Column): TypedColumn[Any, Tile] =
-    Exp10(tileCol)
+  def rf_exp10(tileCol: Column): Column = Exp10(tileCol)
 
   /** Two to the power of cell values */
-  def rf_exp2(tileCol: Column): TypedColumn[Any, Tile] =
-    Exp2(tileCol)
+  def rf_exp2(tileCol: Column): Column = Exp2(tileCol)
 
   /** Exponential of cell values, less one*/
-  def rf_expm1(tileCol: Column): TypedColumn[Any, Tile] =
-    ExpM1(tileCol)
+  def rf_expm1(tileCol: Column): Column = ExpM1(tileCol)
 
   /** Return the incoming tile untouched. */
-  def rf_identity(tileCol: Column): TypedColumn[Any, Tile] =
-    Identity(tileCol)
+  def rf_identity(tileCol: Column): Column = Identity(tileCol)
 
   /** Create a row for each cell in Tile. */
   def rf_explode_tiles(cols: Column*): Column = rf_explode_tiles_sample(1.0, None, cols: _*)

core/src/main/scala/org/locationtech/rasterframes/expressions/localops/Abs.scala

@@ -22,10 +22,9 @@
 package org.locationtech.rasterframes.expressions.localops
 
 import geotrellis.raster.Tile
-import org.apache.spark.sql.catalyst.expressions.{Expression, ExpressionDescription}
+import org.apache.spark.sql.Column
 import org.apache.spark.sql.catalyst.expressions.codegen.CodegenFallback
-import org.apache.spark.sql.{Column, TypedColumn}
-import org.locationtech.rasterframes._
+import org.apache.spark.sql.catalyst.expressions.{Expression, ExpressionDescription}
 import org.locationtech.rasterframes.expressions.{NullToValue, UnaryLocalRasterOp}
 
 @ExpressionDescription(
@@ -45,6 +44,5 @@ case class Abs(child: Expression) extends UnaryLocalRasterOp with NullToValue wi
 }
 
 object Abs {
-  def apply(tile: Column): TypedColumn[Any, Tile] =
-    new Column(Abs(tile.expr)).as[Tile]
+  def apply(tile: Column): Column = new Column(Abs(tile.expr))
 }

core/src/main/scala/org/locationtech/rasterframes/expressions/localops/Add.scala

@@ -22,12 +22,11 @@
 package org.locationtech.rasterframes.expressions.localops
 
 import geotrellis.raster.Tile
+import org.apache.spark.sql.Column
 import org.apache.spark.sql.catalyst.InternalRow
 import org.apache.spark.sql.catalyst.expressions.codegen.CodegenFallback
 import org.apache.spark.sql.catalyst.expressions.{Expression, ExpressionDescription}
 import org.apache.spark.sql.functions.lit
-import org.apache.spark.sql.{Column, TypedColumn}
-import org.locationtech.rasterframes._
 import org.locationtech.rasterframes.expressions.BinaryLocalRasterOp
 import org.locationtech.rasterframes.expressions.DynamicExtractors.tileExtractor
 
@@ -65,9 +64,9 @@ case class Add(left: Expression, right: Expression) extends BinaryLocalRasterOp
   }
 }
 object Add {
-  def apply(left: Column, right: Column): TypedColumn[Any, Tile] =
-    new Column(Add(left.expr, right.expr)).as[Tile]
+  def apply(left: Column, right: Column): Column =
+    new Column(Add(left.expr, right.expr))
 
-  def apply[N: Numeric](tile: Column, value: N): TypedColumn[Any, Tile] =
-    new Column(Add(tile.expr, lit(value).expr)).as[Tile]
+  def apply[N: Numeric](tile: Column, value: N): Column =
+    new Column(Add(tile.expr, lit(value).expr))
 }

core/src/main/scala/org/locationtech/rasterframes/expressions/localops/BiasedAdd.scala

@@ -21,12 +21,11 @@
 
 package org.locationtech.rasterframes.expressions.localops
 import geotrellis.raster.Tile
+import org.apache.spark.sql.Column
 import org.apache.spark.sql.catalyst.InternalRow
 import org.apache.spark.sql.catalyst.expressions.codegen.CodegenFallback
 import org.apache.spark.sql.catalyst.expressions.{Expression, ExpressionDescription}
 import org.apache.spark.sql.functions.lit
-import org.apache.spark.sql.{Column, TypedColumn}
-import org.locationtech.rasterframes._
 import org.locationtech.rasterframes.expressions.BinaryLocalRasterOp
 import org.locationtech.rasterframes.expressions.DynamicExtractors.tileExtractor
 import org.locationtech.rasterframes.util.DataBiasedOp
@@ -66,9 +65,9 @@ case class BiasedAdd(left: Expression, right: Expression) extends BinaryLocalRas
   }
 }
 object BiasedAdd {
-  def apply(left: Column, right: Column): TypedColumn[Any, Tile] =
-    new Column(BiasedAdd(left.expr, right.expr)).as[Tile]
+  def apply(left: Column, right: Column): Column =
+    new Column(BiasedAdd(left.expr, right.expr))
 
-  def apply[N: Numeric](tile: Column, value: N): TypedColumn[Any, Tile] =
-    new Column(BiasedAdd(tile.expr, lit(value).expr)).as[Tile]
+  def apply[N: Numeric](tile: Column, value: N): Column =
+    new Column(BiasedAdd(tile.expr, lit(value).expr))
 }

core/src/main/scala/org/locationtech/rasterframes/expressions/localops/Divide.scala

@@ -21,13 +21,12 @@
 
 package org.locationtech.rasterframes.expressions.localops
 
-import org.locationtech.rasterframes._
-import org.locationtech.rasterframes.expressions.BinaryLocalRasterOp
 import geotrellis.raster.Tile
+import org.apache.spark.sql.Column
 import org.apache.spark.sql.catalyst.expressions.codegen.CodegenFallback
 import org.apache.spark.sql.catalyst.expressions.{Expression, ExpressionDescription}
 import org.apache.spark.sql.functions.lit
-import org.apache.spark.sql.{Column, TypedColumn}
+import org.locationtech.rasterframes.expressions.BinaryLocalRasterOp
 
 @ExpressionDescription(
   usage = "_FUNC_(tile, rhs) - Performs cell-wise division between two tiles or a tile and a scalar.",
@@ -49,9 +48,9 @@ case class Divide(left: Expression, right: Expression) extends BinaryLocalRaster
   override protected def op(left: Tile, right: Int): Tile = left.localDivide(right)
 }
 object Divide {
-  def apply(left: Column, right: Column): TypedColumn[Any, Tile] =
-    new Column(Divide(left.expr, right.expr)).as[Tile]
+  def apply(left: Column, right: Column): Column =
+    new Column(Divide(left.expr, right.expr))
 
-  def apply[N: Numeric](tile: Column, value: N): TypedColumn[Any, Tile] =
-    new Column(Divide(tile.expr, lit(value).expr)).as[Tile]
+  def apply[N: Numeric](tile: Column, value: N): Column =
+    new Column(Divide(tile.expr, lit(value).expr))
 }

core/src/main/scala/org/locationtech/rasterframes/expressions/localops/Equal.scala

@@ -21,13 +21,12 @@
 
 package org.locationtech.rasterframes.expressions.localops
 
-import org.locationtech.rasterframes._
-import org.locationtech.rasterframes.expressions.BinaryLocalRasterOp
 import geotrellis.raster.Tile
-import org.apache.spark.sql.catalyst.expressions.{Expression, ExpressionDescription}
+import org.apache.spark.sql.Column
 import org.apache.spark.sql.catalyst.expressions.codegen.CodegenFallback
+import org.apache.spark.sql.catalyst.expressions.{Expression, ExpressionDescription}
 import org.apache.spark.sql.functions.lit
-import org.apache.spark.sql.{Column, TypedColumn}
+import org.locationtech.rasterframes.expressions.BinaryLocalRasterOp
 
 @ExpressionDescription(
   usage = "_FUNC_(lhs, rhs) - Performs cell-wise equality test between two tiles.",
@@ -48,9 +47,9 @@ case class Equal(left: Expression, right: Expression) extends BinaryLocalRasterO
 }
 
 object Equal {
-  def apply(left: Column, right: Column): TypedColumn[Any, Tile] =
-    new Column(Equal(left.expr, right.expr)).as[Tile]
+  def apply(left: Column, right: Column): Column =
+    new Column(Equal(left.expr, right.expr))
 
-  def apply[N: Numeric](tile: Column, value: N): TypedColumn[Any, Tile] =
-    new Column(Equal(tile.expr, lit(value).expr)).as[Tile]
+  def apply[N: Numeric](tile: Column, value: N): Column =
+    new Column(Equal(tile.expr, lit(value).expr))
 }