[SEDONA-303] Port all Sedona Spark functions to Sedona Flink -- Step 1

docs/api/flink/Function.md

@@ -431,6 +431,60 @@ Input: `MULTILINESTRING((0 0, 10 0, 10 10, 0 10, 0 0),(10 10, 20 10, 20 20, 10 2
 
 Output: `MULTIPOLYGON(((0 0,0 10,10 10,10 0,0 0)),((10 10,10 20,20 20,20 10,10 10)))`
 
+## ST_Centroid
+
+Introduction: Return the centroid point of A
+
+Format: `ST_Centroid (A:geometry)`
+
+Since: `v1.5.0`
+
+Example:
+```sql
+SELECT ST_Centroid(polygondf.countyshape)
+FROM polygondf
+```
+
+## ST_CollectionExtract
+
+Introduction: Returns a homogeneous multi-geometry from a given geometry collection.
+
+The type numbers are:
+1. POINT
+2. LINESTRING
+3. POLYGON
+
+If the type parameter is omitted a multi-geometry of the highest dimension is returned.
+
+Format: `ST_CollectionExtract (A:geometry)`
+
+Format: `ST_CollectionExtract (A:geometry, type:Int)`
+
+Since: `v1.5.0`
+
+Example:
+
+```sql
+WITH test_data as (
+    ST_GeomFromText(
+        'GEOMETRYCOLLECTION(POINT(40 10), POLYGON((0 0, 0 5, 5 5, 5 0, 0 0)))'
+    ) as geom
+)
+SELECT ST_CollectionExtract(geom) as c1, ST_CollectionExtract(geom, 1) as c2
+FROM test_data
+
+```
+
+Result:
+
+```
++----------------------------------------------------------------------------+
+|c1                                        |c2                               |
++----------------------------------------------------------------------------+
+|MULTIPOLYGON(((0 0, 0 5, 5 5, 5 0, 0 0))) |MULTIPOINT(40 10)                |              |
++----------------------------------------------------------------------------+
+```
+
 ## ST_ConcaveHull
 
 Introduction: Return the Concave Hull of polgyon A, with alpha set to pctConvex[0, 1] in the Delaunay Triangulation method, the concave hull will not contain a hole unless allowHoles is set to true
@@ -452,6 +506,20 @@ Input: `Polygon ((0 0, 1 2, 2 2, 3 2, 5 0, 4 0, 3 1, 2 1, 1 0, 0 0))`
 
 Output: `POLYGON ((1 2, 2 2, 3 2, 5 0, 4 0, 1 0, 0 0, 1 2))`
 
+## ST_ConvexHull
+
+Introduction: Return the Convex Hull of polgyon A
+
+Format: `ST_ConvexHull (A:geometry)`
+
+Since: `v1.5.0`
+
+Example:
+```sql
+SELECT ST_ConvexHull(polygondf.countyshape)
+FROM polygondf
+```
+
 ## ST_Dimension
 
 Introduction: Return the topological dimension of this Geometry object, which must be less than or equal to the coordinate dimension. OGC SPEC s2.1.1.1 - returns 0 for POINT, 1 for LINESTRING, 2 for POLYGON, and the largest dimension of the components of a GEOMETRYCOLLECTION. If the dimension is unknown (e.g. for an empty GEOMETRYCOLLECTION) 0 is returned.
@@ -546,6 +614,42 @@ SELECT ST_Degrees(0.19739555984988044)
 
 Output: 11.309932474020195
 
+## ST_Difference
+
+Introduction: Return the difference between geometry A and B (return part of geometry A that does not intersect geometry B)
+
+Format: `ST_Difference (A:geometry, B:geometry)`
+
+Since: `v1.5.0`
+
+Example:
+
+```sql
+SELECT ST_Difference(ST_GeomFromWKT('POLYGON ((-3 -3, 3 -3, 3 3, -3 3, -3 -3))'), ST_GeomFromWKT('POLYGON ((0 -4, 4 -4, 4 4, 0 4, 0 -4))'))
+```
+
+Result:
+
+```
+POLYGON ((0 -3, -3 -3, -3 3, 0 3, 0 -3))
+```
+
+## ST_Dump
+
+Introduction: It expands the geometries. If the geometry is simple (Point, Polygon Linestring etc.) it returns the geometry
+itself, if the geometry is collection or multi it returns record for each of collection components.
+
+Format: `ST_Dump(geom: geometry)`
+
+Since: `v1.5.0`
+
+SQL example:
+```sql
+SELECT ST_Dump(ST_GeomFromText('MULTIPOINT ((10 40), (40 30), (20 20), (30 10))'))
+```
+
+Output: `[POINT (10 40), POINT (40 30), POINT (20 20), POINT (30 10)]`
+
 ## ST_Envelope
 
 Introduction: Return the envelop boundary of A

flink/src/main/java/org/apache/sedona/flink/Catalog.java

@@ -16,6 +16,8 @@
 import org.apache.flink.table.functions.UserDefinedFunction;
 import org.apache.sedona.flink.expressions.*;
 
+import scala.Function;
+
 public class Catalog {
     public static UserDefinedFunction[] getFuncs() {
         return new UserDefinedFunction[]{
@@ -42,13 +44,18 @@ public static UserDefinedFunction[] getFuncs() {
                 new Functions.ST_Azimuth(),
                 new Functions.ST_Boundary(),
                 new Functions.ST_Buffer(),
+                new Functions.ST_Centroid(),
+                new Functions.ST_CollectionExtract(),
                 new Functions.ST_ConcaveHull(),
+                new Functions.ST_ConvexHull(),
                 new Functions.ST_Envelope(),
+                new Functions.ST_Difference(),
                 new Functions.ST_Dimension(),
                 new Functions.ST_Distance(),
                 new Functions.ST_DistanceSphere(),
                 new Functions.ST_DistanceSpheroid(),
                 new Functions.ST_3DDistance(),
+                new Functions.ST_Dump(),
                 new Functions.ST_Length(),
                 new Functions.ST_LengthSpheroid(),
                 new Functions.ST_Transform(),

flink/src/main/java/org/apache/sedona/flink/expressions/Functions.java

@@ -71,6 +71,29 @@ public Geometry eval(@DataTypeHint(value = "RAW", bridgedTo = org.locationtech.j
         }
     }
 
+    public static class ST_Centroid extends ScalarFunction {
+        @DataTypeHint(value = "RAW", bridgedTo = org.locationtech.jts.geom.Geometry.class)
+        public Geometry eval(@DataTypeHint(value = "RAW", bridgedTo = org.locationtech.jts.geom.Geometry.class) Object o) {
+            Geometry geom = (Geometry) o;
+            return org.apache.sedona.common.Functions.getCentroid(geom);
+        }
+    }
+
+    public static class ST_CollectionExtract extends ScalarFunction {
+        @DataTypeHint(value = "RAW", bridgedTo = org.locationtech.jts.geom.Geometry.class)
+        public Geometry eval(@DataTypeHint(value = "RAW", bridgedTo = org.locationtech.jts.geom.Geometry.class) Object o) {
+            Geometry geom = (Geometry) o;
+            return org.apache.sedona.common.Functions.collectionExtract(geom);
+        }
+
+        @DataTypeHint(value = "RAW", bridgedTo = org.locationtech.jts.geom.Geometry.class)
+        public Geometry eval(@DataTypeHint(value = "RAW", bridgedTo = org.locationtech.jts.geom.Geometry.class) Object o,
+                             @DataTypeHint("Integer") Integer geoType) {
+            Geometry geom = (Geometry) o;
+            return org.apache.sedona.common.Functions.collectionExtract(geom, geoType);
+        }
+    }
+
     public static class ST_ConcaveHull extends ScalarFunction {
         @DataTypeHint(value = "RAW", bridgedTo = org.locationtech.jts.geom.Geometry.class)
         public Geometry eval(@DataTypeHint(value = "RAW", bridgedTo = org.locationtech.jts.geom.Geometry.class) Object o,
@@ -87,6 +110,14 @@ public Geometry eval(@DataTypeHint(value = "RAW", bridgedTo = org.locationtech.j
         }
     }
 
+    public static class ST_ConvexHull extends ScalarFunction {
+        @DataTypeHint(value = "RAW", bridgedTo = org.locationtech.jts.geom.Geometry.class)
+        public Geometry eval(@DataTypeHint(value = "RAW", bridgedTo = org.locationtech.jts.geom.Geometry.class) Object o) {
+            Geometry geom = (Geometry) o;
+            return org.apache.sedona.common.Functions.convexHull(geom);
+        }
+    }
+
     public static class ST_Envelope extends ScalarFunction {
         @DataTypeHint(value = "RAW", bridgedTo = org.locationtech.jts.geom.Geometry.class)
         public Geometry eval(@DataTypeHint(value = "RAW", bridgedTo = org.locationtech.jts.geom.Geometry.class) Object o) {
@@ -101,7 +132,16 @@ public Integer eval(@DataTypeHint(value = "RAW", bridgedTo = org.locationtech.jt
             Geometry geom = (Geometry) o;
             return org.apache.sedona.common.Functions.dimension(geom);
         }
+    }
 
+    public static class ST_Difference extends ScalarFunction {
+        @DataTypeHint(value = "RAW", bridgedTo = org.locationtech.jts.geom.Geometry.class)
+        public Geometry eval(@DataTypeHint(value = "RAW", bridgedTo = org.locationtech.jts.geom.Geometry.class) Object o1,
+                           @DataTypeHint(value = "RAW", bridgedTo = org.locationtech.jts.geom.Geometry.class) Object o2) {
+            Geometry geom1 = (Geometry) o1;
+            Geometry geom2 = (Geometry) o2;
+            return org.apache.sedona.common.Functions.difference(geom1, geom2);
+        }
     }
 
     public static class ST_Distance extends ScalarFunction {
@@ -152,6 +192,14 @@ public Double eval(@DataTypeHint(value = "RAW", bridgedTo = org.locationtech.jts
         }
     }
 
+    public static class ST_Dump extends ScalarFunction {
+        @DataTypeHint(value = "RAW", bridgedTo = org.locationtech.jts.geom.Geometry[].class)
+        public Geometry[] eval(@DataTypeHint(value = "RAW", bridgedTo = org.locationtech.jts.geom.Geometry.class) Object o) {
+            Geometry geom1 = (Geometry) o;
+            return org.apache.sedona.common.Functions.dump(geom1);
+        }
+    }
+
     public static class ST_Length extends ScalarFunction {
         @DataTypeHint("Double")
         public Double eval(@DataTypeHint(value = "RAW", bridgedTo = org.locationtech.jts.geom.Geometry.class) Object o) {

flink/src/test/java/org/apache/sedona/flink/FunctionTest.java

@@ -84,6 +84,22 @@ public void testBuffer() {
         assert(result instanceof Polygon);
     }
 
+    @Test
+    public void testCentroid() {
+        Table polygonTable = tableEnv.sqlQuery("SELECT ST_GeomFromText('POLYGON ((2 2, 0 0, 2 0, 0 2, 2 2))') as geom");
+        Table resultTable = polygonTable.select(call(Functions.ST_Centroid.class.getSimpleName(), $("geom")));
+        Geometry result = (Geometry) first(resultTable).getField(0);
+        assertEquals("POINT (1 1)", result.toString());
+    }
+
+    @Test
+    public void testCollectionExtract() {
+        Table collectionTable = tableEnv.sqlQuery("SELECT ST_GeomFromText('GEOMETRYCOLLECTION(POINT(0 0), LINESTRING(1 1, 2 2))') as collection");
+        Table resultTable = collectionTable.select(call(Functions.ST_CollectionExtract.class.getSimpleName(), $("collection")));
+        Geometry result = (Geometry) first(resultTable).getField(0);
+        assertEquals("MULTILINESTRING ((1 1, 2 2))", result.toString());
+    }
+
     @Test
     public void testConcaveHull() {
         Table polygonTable = tableEnv.sqlQuery("SELECT ST_GeomFromWKT('Polygon ((0 0, 1 2, 2 2, 3 2, 5 0, 4 0, 3 1, 2 1, 1 0, 0 0))') as geom");
@@ -97,6 +113,31 @@ public void testConcaveHull() {
         assertEquals("POLYGON ((0 0, 1 1, 1 0, 0 0))", result2.toString());
     }
 
+    @Test
+    public void testConvexHull() {
+        Table polygonTable = tableEnv.sqlQuery("SELECT ST_GeomFromWKT('Polygon ((0 0, 1 2, 2 2, 3 2, 5 0, 4 0, 3 1, 2 1, 1 0, 0 0))') as geom");
+        Table concaveHullPolygonTable = polygonTable.select(call(Functions.ST_ConvexHull.class.getSimpleName(), $("geom")));
+        Geometry result = (Geometry) first(concaveHullPolygonTable).getField(0);
+        assertEquals("POLYGON ((0 0, 1 2, 3 2, 5 0, 0 0))", result.toString());
+    }
+
+    @Test
+    public void testDifference() {
+        Table lineTable = tableEnv.sqlQuery("SELECT ST_GeomFromWKT('LINESTRING(50 100, 50 200)') AS g1, ST_GeomFromWKT('LINESTRING(50 50, 50 150)') as g2");
+        Table resultTable = lineTable.select(call(Functions.ST_Difference.class.getSimpleName(), $("g1"), $("g2")));
+        Geometry result = (Geometry) first(resultTable).getField(0);
+        assertEquals("LINESTRING (50 150, 50 200)", result.toString());
+    }
+
+    @Test
+    public void testDump() {
+        Table table = tableEnv.sqlQuery("SELECT ST_GeomFromWKT('MULTIPOINT ((0 0), (1 1))') AS geom");
+        table = table.select(call(Functions.ST_Dump.class.getSimpleName(), $("geom")));
+        Geometry[] result = (Geometry[]) first(table).getField(0);
+        assertEquals("POINT (0 0)", result[0].toString());
+        assertEquals("POINT (1 1)", result[1].toString());
+    }
+
     @Test
     public void testEnvelope() {
         Table linestringTable = createLineStringTable(1);