GEOMESA-3519 Spark - Update to Sedona 1.8.0

Author: elahrvivaz

build/cqs.tsv

@@ -318,8 +318,8 @@ org.apache.htrace:htrace-core4	4.1.0-incubating	provided
 org.apache.kafka:kafka-clients	3.9.0	provided
 org.apache.kafka:kafka-streams-scala_2.12	3.9.0	provided
 org.apache.kafka:kafka-streams	3.9.0	provided
-org.apache.sedona:sedona-spark-3.4_2.12	1.5.0	provided
-org.apache.sedona:sedona-spark-common-3.4_2.12	1.5.0	provided
+org.apache.sedona:sedona-spark-3.5_2.12	1.8.0	provided
+org.apache.sedona:sedona-spark-common-3.5_2.12	1.8.0	provided
 org.apache.spark:spark-catalyst_2.12	3.5.5	provided
 org.apache.spark:spark-core_2.12	3.5.5	provided
 org.apache.spark:spark-sql_2.12	3.5.5	provided
@@ -338,7 +338,7 @@ org.apache.arrow:arrow-vector	tests:18.3.0	test
 org.apache.curator:curator-test	5.6.0	test
 org.apache.kafka:kafka-clients	test:3.9.0	test
 org.apache.kafka:kafka-streams-test-utils	3.9.0	test
-org.apache.sedona:sedona-common	1.5.0	test
+org.apache.sedona:sedona-common	1.8.0	test
 org.codehaus.groovy:groovy-jsr223	3.0.20	test
 org.duckdb:duckdb_jdbc	1.2.2.0	test
 org.geomesa.testcontainers:testcontainers-accumulo	1.4.1	test

docs/user/upgrade.rst

@@ -107,6 +107,7 @@ The following dependencies have been upgraded:
 * nifi ``2.4.0`` -> ``2.5.0``
 * orc ``1.9.6`` -> ``1.9.7``
 * postgresql ``42.7.2`` -> ``42.7.7``
+* sedona ``1.5.0`` -> ``1.8.0``
 
 Switch to Micrometer Metrics
 ----------------------------

geomesa-spark/geomesa-spark-jts/src/main/scala/org/locationtech/geomesa/spark/jts/package.scala

@@ -20,8 +20,12 @@ import scala.util.Try
  */
 package object jts extends DataFrameFunctions.Library with SpatialEncoders {
 
-  lazy val SedonaGeometryUDT: Try[UserDefinedType[Geometry]] =
-    Try(Class.forName("org.apache.spark.sql.sedona_sql.UDT.GeometryUDT").newInstance().asInstanceOf[UserDefinedType[Geometry]])
+  lazy val SedonaGeometryUDT: Try[UserDefinedType[Geometry]] = Try {
+    Class.forName("org.apache.spark.sql.sedona_sql.UDT.GeometryUDT")
+      .getDeclaredConstructor()
+      .newInstance()
+      .asInstanceOf[UserDefinedType[Geometry]]
+  }
 
   def useSedonaSerialization: Boolean =
     sys.props.get("geomesa.use.sedona").forall(_.toBoolean) && SedonaGeometryUDT.isSuccess

geomesa-spark/geomesa-spark-sql/src/main/scala/org/locationtech/geomesa/spark/sql/SQLRules.scala

@@ -15,7 +15,7 @@ import org.apache.spark.sql.catalyst.rules.Rule
 import org.apache.spark.sql.execution.datasources.LogicalRelation
 import org.apache.spark.sql.execution.{ProjectExec, SparkPlan}
 import org.apache.spark.sql.sedona_sql.UDT.{GeometryUDT => Sedona_GeometryUDT}
-import org.apache.spark.sql.sedona_sql.expressions.{ST_Contains => Sedona_ST_Contains, ST_Crosses => Sedona_ST_Crosses, ST_Equals => Sedona_ST_Equals, ST_Intersects => Sedona_ST_Intersects, ST_Overlaps => Sedona_ST_Overlaps, ST_Predicate => Sedona_ST_Predicate, ST_Touches => Sedona_ST_Touches, ST_Within => Sedona_ST_Within}
+import org.apache.spark.sql.sedona_sql.expressions.{ST_Predicate => Sedona_ST_Predicate}
 import org.apache.spark.sql.types.DataTypes
 import org.apache.spark.sql.{SQLContext, Strategy}
 import org.geotools.api.filter.expression.{Expression => GTExpression, Literal => GTLiteral}
@@ -74,20 +74,21 @@ object SQLRules extends LazyLogging {
   }
 
   def sedonaExprToGTFilter(pred: Sedona_ST_Predicate): Option[GTFilter] = {
-    val left = pred.children.head
-    val right = pred.children.last
-    (sparkExprToGTExpr(left), sparkExprToGTExpr(right)) match {
-      case (Some(expr1), Some(expr2)) => pred match {
-        case Sedona_ST_Contains(_) => Some(ff.contains(expr1, expr2))
-        case Sedona_ST_Crosses(_) => Some(ff.crosses(expr1, expr2))
-        case Sedona_ST_Overlaps(_) => Some(ff.overlaps(expr1, expr2))
-        case Sedona_ST_Intersects(_) => Some(ff.intersects(expr1, expr2))
-        case Sedona_ST_Within(_) => Some(ff.within(expr1, expr2))
-        case Sedona_ST_Touches(_) => Some(ff.touches(expr1, expr2))
-        case Sedona_ST_Equals(_) => Some(ff.equal(expr1, expr2))
-        case _ => None
+    sparkExprToGTExpr(pred.children.head).flatMap { expr1 =>
+      sparkExprToGTExpr(pred.children.last).flatMap { expr2 =>
+        // sedona classes are private, so we have to match on the class name instead of the class itself
+        pred.getClass.getSimpleName match {
+          case "ST_Contains"   => Some(ff.contains(expr1, expr2))
+          case "ST_Crosses"    => Some(ff.crosses(expr1, expr2))
+          case "ST_Overlaps"   => Some(ff.overlaps(expr1, expr2))
+          case "ST_Intersects" => Some(ff.intersects(expr1, expr2))
+          case "ST_Within"     => Some(ff.within(expr1, expr2))
+          case "ST_Touches"    => Some(ff.touches(expr1, expr2))
+          case "ST_Equals"     => Some(ff.equal(expr1, expr2))
+          case "ST_Disjoint"   => Some(ff.disjoint(expr1, expr2))
+          case _ => None
+        }
       }
-      case _ => None
     }
   }
 

pom.xml

@@ -145,8 +145,8 @@
         <kafka.version>3.9.0</kafka.version> <!-- needs to align with confluent version -->
         <confluent.version>7.8.0</confluent.version> <!-- confluent 7.8.x corresponds to kafka 3.8.x -->
         <confluent.protobuf.version>3.25.6</confluent.protobuf.version>
-        <sedona.version>1.5.0</sedona.version>
-        <sedona.spark.version>3.4</sedona.spark.version>
+        <sedona.version>1.8.0</sedona.version>
+        <sedona.spark.version>3.5</sedona.spark.version>
         <sedona.h3.version>4.1.1</sedona.h3.version>
         <jts2geojson.version>0.16.1</jts2geojson.version> <!-- for testing Apache Sedona integration -->
         <snappy.java.version>1.1.10.5</snappy.java.version>