Fixes to UserDefinedAggregateFunction implementations so they run in the Databricks environment.

Not sure what the cause is, but this change set seems to fix it.

Author: metasim

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

@@ -123,8 +123,8 @@ object CellStatsAggregate {
   import org.locationtech.rasterframes.encoders.StandardEncoders.cellStatsEncoder
 
   def apply(col: Column): TypedColumn[Any, CellStatistics] =
-    new Column(new CellStatsAggregateUDAF(col.expr))
-      .as(s"rf_agg_stats($col)") // node renaming in class doesn't seem to propagate
+    new CellStatsAggregate()(ExtractTile(col))
+      .as(s"rf_agg_stats($col)")
       .as[CellStatistics]
 
   /** Adapter hack to allow UserDefinedAggregateFunction to be referenced as an expression. */

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

@@ -98,8 +98,8 @@ object HistogramAggregate {
   import org.locationtech.rasterframes.encoders.StandardEncoders.cellHistEncoder
 
   def apply(col: Column): TypedColumn[Any, CellHistogram] =
-    new Column(new HistogramAggregateUDAF(col.expr))
-      .as(s"rf_agg_approx_histogram($col)") // node renaming in class doesn't seem to propogate
+    new HistogramAggregate()(ExtractTile(col))
+      .as(s"rf_agg_approx_histogram($col)")
       .as[CellHistogram]
 
   /** Adapter hack to allow UserDefinedAggregateFunction to be referenced as an expression. */

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

@@ -92,7 +92,7 @@ object LocalCountAggregate {
   object LocalDataCellsUDAF {
     def apply(child: Expression): LocalDataCellsUDAF = new LocalDataCellsUDAF(child)
     def apply(tile: Column): TypedColumn[Any, Tile] =
-      new Column(new LocalDataCellsUDAF(tile.expr))
+      new LocalCountAggregate(true)(ExtractTile(tile))
         .as(s"rf_agg_local_data_cells($tile)")
         .as[Tile]
   }
@@ -107,7 +107,7 @@ object LocalCountAggregate {
   object LocalNoDataCellsUDAF {
     def apply(child: Expression): LocalNoDataCellsUDAF = new LocalNoDataCellsUDAF(child)
     def apply(tile: Column): TypedColumn[Any, Tile] =
-      new Column(new LocalNoDataCellsUDAF(tile.expr))
+      new LocalCountAggregate(false)(ExtractTile(tile))
         .as(s"rf_agg_local_no_data_cells($tile)")
         .as[Tile]
   }

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

@@ -146,7 +146,7 @@ class LocalStatsAggregate() extends UserDefinedAggregateFunction {
 object LocalStatsAggregate {
 
   def apply(col: Column): TypedColumn[Any, LocalCellStatistics] =
-    new Column(LocalStatsAggregateUDAF(col.expr))
+    new LocalStatsAggregate()(ExtractTile(col))
       .as(s"rf_agg_local_stats($col)")
       .as[LocalCellStatistics]
 

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

@@ -83,7 +83,10 @@ object LocalTileOpAggregate {
   }
   object LocalMinUDAF {
     def apply(child: Expression): LocalMinUDAF = new LocalMinUDAF(child)
-    def apply(tile: Column): TypedColumn[Any, Tile] = new Column(new LocalMinUDAF(tile.expr)).as[Tile]
+    def apply(tile: Column): TypedColumn[Any, Tile] =
+      new LocalTileOpAggregate(BiasedMin)(ExtractTile(tile))
+        .as(s"rf_agg_local_min($tile)")
+        .as[Tile]
   }
 
   @ExpressionDescription(
@@ -95,6 +98,9 @@ object LocalTileOpAggregate {
   }
   object LocalMaxUDAF {
     def apply(child: Expression): LocalMaxUDAF = new LocalMaxUDAF(child)
-    def apply(tile: Column): TypedColumn[Any, Tile] = new Column(new LocalMaxUDAF(tile.expr)).as[Tile]
+    def apply(tile: Column): TypedColumn[Any, Tile] =
+      new LocalTileOpAggregate(BiasedMax)(ExtractTile(tile))
+        .as(s"rf_agg_local_max($tile)")
+        .as[Tile]
   }
 }