[SEDONA-743] Add geom from mysql function, to support loading MySQL geometry object directly to Sedona. (#2287)

* SEDONA-738 Fix unit tests.

* SEDONA-738 Fix unit tests.

* SEDONA-738 Fix unit tests.

* SEDONA-738 Fix unit tests.

* SEDONA-738 Fix unit tests.

* SEDONA-738 Fix unit tests.

* SEDONA-743 Add geom from mysql function.

* SEDONA-743 Add geom from mysql function.

* SEDONA-743 Add geom from mysql function.

* SEDONA-743 Add geom from mysql function.

* SEDONA-743 Add geom from mysql function.

* Update docs/api/sql/Constructor.md

---------

Co-authored-by: Jia Yu <[email protected]>

Author: Imbruced

common/src/main/java/org/apache/sedona/common/Constructors.java

@@ -19,6 +19,8 @@
 package org.apache.sedona.common;
 
 import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.nio.ByteOrder;
 import javax.xml.parsers.ParserConfigurationException;
 import org.apache.sedona.common.enums.FileDataSplitter;
 import org.apache.sedona.common.enums.GeometryType;
@@ -302,4 +304,21 @@ public static Geometry geomFromGML(String gml)
   public static Geometry geomFromKML(String kml) throws ParseException {
     return new KMLReader().read(kml);
   }
+
+  public static Geometry geomFromMySQL(byte[] binary) throws ParseException {
+    ByteBuffer buffer = ByteBuffer.wrap(binary);
+
+    buffer.order(ByteOrder.LITTLE_ENDIAN);
+    int srid = buffer.getInt();
+
+    byte[] wkb = new byte[buffer.remaining()];
+
+    buffer.get(wkb);
+
+    Geometry geom = geomFromWKB(wkb);
+
+    geom.setSRID(srid);
+
+    return geom;
+  }
 }

docs/api/sql/Constructor.md

@@ -842,3 +842,32 @@ Output:
 ```
 POLYGON ((-74.0428197 40.6867969, -74.0421975 40.6921336, -74.050802 40.6912794, -74.0428197 40.6867969))
 ```
+
+## ST_GeomFromMySQL
+
+Introduction: Construct a Geometry from MySQL Geometry binary.
+
+Format: `ST_GeomFromMySQL (binary: Binary)`
+
+Since: `v1.8.0`
+
+SQL Example
+
+```sql
+SELECT
+    ST_GeomFromMySQL(geomWKB) AS geom,
+    ST_SRID(ST_GeomFromMySQL(geomWKB)) AS srid
+FROM mysql_table
+```
+
+Output:
+
+```
++-------------+----+
+|         geom|srid|
++-------------+----+
+|POINT (20 10)|4326|
+|POINT (40 30)|4326|
+|POINT (60 50)|4326|
++-------------+----+
+```

spark/common/pom.xml

@@ -259,6 +259,18 @@
             <version>1.20.0</version>
             <scope>test</scope>
         </dependency>
+        <dependency>
+            <groupId>org.testcontainers</groupId>
+            <artifactId>mysql</artifactId>
+            <version>1.20.0</version>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>com.mysql</groupId>
+            <artifactId>mysql-connector-j</artifactId>
+            <version>8.4.0</version>
+            <scope>test</scope>
+        </dependency>
         <dependency>
             <groupId>io.minio</groupId>
             <artifactId>minio</artifactId>

spark/common/src/main/scala/org/apache/sedona/sql/UDF/Catalog.scala

@@ -215,6 +215,7 @@ object Catalog extends AbstractCatalog {
     function[ST_MLineFromText](0),
     function[ST_GeomCollFromText](0),
     function[ST_GeogCollFromText](0),
+    function[ST_GeomFromMySQL](),
     function[ST_Split](),
     function[ST_S2CellIDs](),
     function[ST_S2ToGeom](),

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

@@ -624,3 +624,11 @@ private[apache] case class ST_GeomCollFromText(inputExpressions: Seq[Expression]
     copy(inputExpressions = newChildren)
   }
 }
+
+private[apache] case class ST_GeomFromMySQL(inputExpressions: Seq[Expression])
+    extends InferredExpression(Constructors.geomFromMySQL _) {
+
+  protected def withNewChildrenInternal(newChildren: IndexedSeq[Expression]) = {
+    copy(inputExpressions = newChildren)
+  }
+}

spark/common/src/test/resources/mysql/init_mysql.sql

@@ -0,0 +1,17 @@
+CREATE TABLE points
+(
+    name     VARCHAR(50),
+    location GEOMETRY
+);
+
+INSERT INTO points (name, location)
+VALUES ('Point A',
+        ST_GeomFromText('POINT(10 20)', 4326));
+
+INSERT INTO points (name, location)
+VALUES ('Point B',
+        ST_GeomFromText('POINT(30 40)', 4326));
+
+INSERT INTO points (name, location)
+VALUES ('Point C',
+        ST_GeomFromText('POINT(50 60)', 4326));

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

@@ -21,9 +21,12 @@ package org.apache.sedona.sql
 import org.apache.sedona.core.formatMapper.GeoJsonReader
 import org.apache.sedona.core.formatMapper.shapefileParser.ShapefileReader
 import org.apache.sedona.sql.utils.Adapter
+import org.apache.spark.sql.Row
 import org.locationtech.jts.geom.{Geometry, LineString}
+import org.scalatest.matchers.should.Matchers
+import org.testcontainers.containers.MySQLContainer
 
-class constructorTestScala extends TestBaseScala {
+class constructorTestScala extends TestBaseScala with Matchers {
 
   import sparkSession.implicits._
 
@@ -626,5 +629,49 @@ class constructorTestScala extends TestBaseScala {
       val actualSrid = actualGeom.getSRID
       assert(4326 == actualSrid)
     }
+
+    it("should properly read data from MySQL") {
+      val runTest = (jdbcURL: String) => {
+        val tableName = "points"
+        val properties = new java.util.Properties()
+        properties.setProperty("user", "sedona")
+        properties.setProperty("password", "sedona")
+
+        sparkSession.read
+          .jdbc(jdbcURL, tableName, properties)
+          .selectExpr(
+            "ST_GeomFromMySQL(location) as geom",
+            "ST_SRID(ST_GeomFromMySQL(location)) AS srid")
+          .show
+
+        val elements = sparkSession.read
+          .jdbc(jdbcURL, tableName, properties)
+          .selectExpr(
+            "ST_GeomFromMySQL(location) as geom",
+            "ST_SRID(ST_GeomFromMySQL(location)) AS srid")
+          .selectExpr("ST_AsText(geom) as geom", "srid")
+          .collect()
+
+        elements.length shouldBe 3
+        elements should contain theSameElementsAs Seq(
+          Row("POINT (20 10)", 4326),
+          Row("POINT (40 30)", 4326),
+          Row("POINT (60 50)", 4326))
+      }
+
+      val mysql = new MySQLContainer("mysql:8.4.0")
+      mysql.withInitScript("mysql/init_mysql.sql")
+      mysql.withUsername("sedona")
+      mysql.withPassword("sedona")
+      mysql.withDatabaseName("sedona")
+
+      mysql.start()
+
+      try {
+        runTest(mysql.getJdbcUrl)
+      } finally {
+        mysql.stop()
+      }
+    }
   }
 }