[SEDONA-630] Fix the ENCODER_NOT_FOUND error for ST_Union_Aggr output…

…Encoder (#1545)

Always construct the encoders on the driver. If encoders are constructed on the executor then Spark will complain that it doesn't know how to encode Geometry.

spark/common/src/main/scala/org/apache/spark/sql/sedona_sql/expressions/AggregateFunctions.scala

@@ -55,8 +55,10 @@ trait TraitSTAggregateExec {
 }
 
 class ST_Union_Aggr(bufferSize: Int = 1000)
-    extends Aggregator[Geometry, ListBuffer[Geometry], Geometry]
-    with Serializable {
+    extends Aggregator[Geometry, ListBuffer[Geometry], Geometry] {
+
+  val serde = ExpressionEncoder[Geometry]()
+  val bufferSerde = ExpressionEncoder[ListBuffer[Geometry]]()
 
   override def reduce(buffer: ListBuffer[Geometry], input: Geometry): ListBuffer[Geometry] = {
     buffer += input
@@ -86,10 +88,9 @@ class ST_Union_Aggr(bufferSize: Int = 1000)
     OverlayNGRobust.union(reduction.asJava)
   }
 
-  def bufferEncoder: ExpressionEncoder[ListBuffer[Geometry]] =
-    ExpressionEncoder[ListBuffer[Geometry]]()
+  def bufferEncoder: ExpressionEncoder[ListBuffer[Geometry]] = bufferSerde
 
-  def outputEncoder: ExpressionEncoder[Geometry] = ExpressionEncoder[Geometry]()
+  def outputEncoder: ExpressionEncoder[Geometry] = serde
 
   override def zero: ListBuffer[Geometry] = ListBuffer.empty
 }

spark/common/src/test/scala/org/apache/sedona/sql/aggregateFunctionTestScala.scala

@@ -19,10 +19,7 @@
 package org.apache.sedona.sql
 
 import org.apache.spark.sql.DataFrame
-import org.apache.spark.sql.expressions.javalang.typed
-import org.apache.spark.sql.sedona_sql.expressions.ST_Union_Aggr
 import org.locationtech.jts.geom.{Coordinate, Geometry, GeometryFactory, Polygon}
-import org.locationtech.jts.io.WKTReader
 
 import scala.util.Random