[SEDONA-105] Add ST_PointOnSurface function

core/src/main/java/org/apache/sedona/core/utils/GeomUtils.java

@@ -71,4 +71,13 @@ public boolean isDone() {
             }
         });
     }
+    /*
+     * Returns a POINT that is guaranteed to lie on the surface.
+     */
+    public static Geometry getInteriorPoint(Geometry geometry) {
+        if(geometry==null) {
+            return null;
+        }
+        return geometry.getInteriorPoint();
+    }
 }

docs/api/flink/Function.md

@@ -101,6 +101,38 @@ Result:
 +-----------------------------+
 ```
 
+## ST_PointOnSurface
+
+Introduction: Returns a POINT guaranteed to lie on the surface.
+
+Format: `ST_PointOnSurface(A:geometry)`
+
+Since: `v1.2.1`
+
+Examples: 
+
+
+```SQL
+SELECT ST_PointOnSurface(df.geometry)
+FROM df
+```
+
+1.  Input: `POINT (0 5)`
+
+    Output: `POINT (0 5)`
+
+2.  Input: `LINESTRING(0 5, 0 10)`
+
+    Output: `POINT (0 5)`
+
+3.  Input: `POLYGON((0 0, 0 5, 5 5, 5 0, 0 0))`
+
+    Output: `POINT (2.5 2.5)`
+
+4.  Input: `LINESTRING(0 5 1, 0 0 1, 0 10 2)`
+
+    Output: `POINT Z(0 0 1)`  
+
 ## ST_Reverse
 
 Introduction: Return the geometry with vertex order reversed

docs/api/sql/Function.md

@@ -1030,6 +1030,39 @@ Result:
 POLYGON ((3 -1, 3 -3, -3 -3, -3 3, 3 3, 3 1, 5 0, 3 -1))
 ```
 
+## ST_PointOnSurface
+
+Introduction: Returns a POINT guaranteed to lie on the surface.
+
+Format: `ST_PointOnSurface(A:geometry)`
+
+Since: `v1.2.1`
+
+Examples: 
+
+```
+SELECT ST_AsText(ST_PointOnSurface(ST_GeomFromText('POINT(0 5)')));
+ st_astext
+------------
+ POINT(0 5)
+
+SELECT ST_AsText(ST_PointOnSurface(ST_GeomFromText('LINESTRING(0 5, 0 10)')));
+ st_astext
+------------
+ POINT(0 5)
+
+SELECT ST_AsText(ST_PointOnSurface(ST_GeomFromText('POLYGON((0 0, 0 5, 5 5, 5 0, 0 0))')));
+   st_astext
+----------------
+ POINT(2.5 2.5)
+
+SELECT ST_AsText(ST_PointOnSurface(ST_GeomFromText('LINESTRING(0 5 1, 0 0 1, 0 10 2)')));
+   st_astext
+----------------
+ POINT Z(0 0 1)
+
+```
+
 ## ST_Reverse
 
 Introduction: Return the geometry with vertex order reversed

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

@@ -31,6 +31,7 @@ public static UserDefinedFunction[] getFuncs() {
                 new Functions.ST_Transform(),
                 new Functions.ST_FlipCoordinates(),
                 new Functions.ST_GeoHash(),
+                new Functions.ST_PointOnSurface(),
                 new Functions.ST_Reverse()
         };
     }

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

@@ -85,6 +85,15 @@ public Optional<String> eval(@DataTypeHint(value = "RAW", bridgedTo = org.locati
         }
     }
 
+    public static class ST_PointOnSurface 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;
+            GeomUtils.getInteriorPoint(geom);
+            return geom;
+        }
+    }
+
     public static class ST_Reverse 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) {

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

@@ -66,6 +66,13 @@ public void testGeomToGeoHash() {
     }
 
     @Test
+    public void testPointOnSurface() {
+        Table pointTable = createPointTable_real(testDataSize);
+        Table surfaceTable = pointTable.select(call(Functions.ST_PointOnSurface.class.getSimpleName(), $(pointColNames[0])));
+        Geometry result = (Geometry) first(surfaceTable).getField(0);
+        assertEquals("POINT (32 -118)", result.toString());
+    }
+
     public void testReverse() {
         Table polygonTable = createPolygonTable(1);
         Table ReversedTable = polygonTable.select(call(Functions.ST_Reverse.class.getSimpleName(), $(polygonColNames[0])));

python/tests/sql/test_function.py

@@ -935,4 +935,25 @@ def __wkt_list_to_data_frame(self, wkt_list: List) -> DataFrame:
         return self.spark.createDataFrame([[wkt.loads(given_wkt)] for given_wkt in wkt_list], self.geo_schema)
 
     def __wkt_pair_list_with_index_to_data_frame(self, wkt_list: List) -> DataFrame:
-        return self.spark.createDataFrame([[index, wkt.loads(given_wkt)] for index, given_wkt in wkt_list], self.geo_schema_with_index)
+        return self.spark.createDataFrame([[index, wkt.loads(given_wkt)] for index, given_wkt in wkt_list], self.geo_schema_with_index)
+
+    def test_st_pointonsurface(self):
+        tests1 = {
+        "'POINT(0 5)'":"POINT (0 5)",
+        "'LINESTRING(0 5, 0 10)'":"POINT (0 5)",
+        "'POLYGON((0 0, 0 5, 5 5, 5 0, 0 0))'":"POINT (2.5 2.5)",
+        "'LINESTRING(0 5 1, 0 0 1, 0 10 2)'":"POINT Z(0 0 1)"
+        }
+
+        for input_geom, expected_geom in tests1.items():
+            pointOnSurface = self.spark.sql("select ST_AsText(ST_PointOnSurface(ST_GeomFromText({})))".format(input_geom))
+            assert pointOnSurface.take(1)[0][0] == expected_geom
+
+        '''
+        ST_AsEWKT Has not been implemented yet
+        tests2 = { "'LINESTRING(0 5 1, 0 0 1, 0 10 2)'":"POINT(0 0 1)" }
+
+        for input_geom, expected_geom in tests2.items():
+            pointOnSurface = self.spark.sql("select ST_AsEWKT(ST_PointOnSurface(ST_GeomFromEWKT({})))".format(input_geom))
+            assert pointOnSurface.take(1)[0][0] == expected_geom
+        '''

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

@@ -101,6 +101,7 @@ object Catalog {
     ST_GeomFromGeoHash,
     ST_Collect,
     ST_Multi,
+    ST_PointOnSurface,
     ST_Reverse,
     // Expression for rasters
     RS_NormalizedDifference,

sql/src/main/scala/org/apache/spark/sql/sedona_sql/expressions/Functions.scala

@@ -1529,6 +1529,28 @@ case class ST_Multi(inputExpressions: Seq[Expression]) extends UnaryGeometryExpr
   }
 }
 
+/**
+ * Returns a POINT guaranteed to lie on the surface.
+ *
+ * @param inputExpressions Geometry
+ */
+case class ST_PointOnSurface(inputExpressions: Seq[Expression])
+  extends UnaryGeometryExpression with CodegenFallback {
+  assert(inputExpressions.length == 1)
+
+  override protected def nullSafeEval(geometry: Geometry): Any = {
+    new GenericArrayData(GeometrySerializer.serialize(GeomUtils.getInteriorPoint(geometry)))
+  }
+
+  override def dataType: DataType = GeometryUDT
+
+  override def children: Seq[Expression] = inputExpressions
+
+  protected def withNewChildrenInternal(newChildren: IndexedSeq[Expression]) = {
+    copy(inputExpressions = newChildren)
+  }
+}
+
 /**
  * Returns the geometry with vertex order reversed
  *

sql/src/test/scala/org/apache/sedona/sql/functionTestScala.scala

@@ -1266,6 +1266,40 @@ class functionTestScala extends TestBaseScala with Matchers with GeometrySample
 
   }
 
+  it ("Should pass ST_PointOnSurface") {
+
+    val geomTestCases1 = Map(
+      "'POINT(0 5)'"
+        -> "POINT (0 5)",
+      "'LINESTRING(0 5, 0 10)'"
+        -> "POINT (0 5)",
+      "'POLYGON((0 0, 0 5, 5 5, 5 0, 0 0))'"
+        -> "POINT (2.5 2.5)",
+      "'LINESTRING(0 5 1, 0 0 1, 0 10 2)'"
+        -> "POINT Z(0 0 1)"
+    )
+
+        for((inputGeom, expectedGeom) <- geomTestCases1) {
+      var df = sparkSession.sql(s"select ST_AsText(ST_PointOnSurface(ST_GeomFromText($inputGeom)))")
+      var result = df.collect()
+      assert(result.head.get(0).asInstanceOf[String]==expectedGeom)
+    }
+
+    /* ST_AsEWKT Has not been implemented yet
+
+    val geomTestCases2 = Map(
+      "'LINESTRING(0 5 1, 0 0 1, 0 10 2)'"
+        -> "POINT (0 0 1)"
+    )
+
+    for((inputGeom, expectedGeom) <- geomTestCases2) {
+      var df = sparkSession.sql(s"select ST_AsEWKT(ST_PointOnSurface(ST_GeomFromEWKT($inputGeom)))")
+      var result = df.collect()
+      assert(result.head.get(0).asInstanceOf[String]==expectedGeom)
+    }
+    */
+  }
+
   it ("Should pass ST_Reverse") {
     val geomTestCases = Map(
       "'POLYGON((-1 0 0, 1 0 0, 0 0 1, 0 1 0, -1 0 0))'"
@@ -1408,6 +1442,8 @@ class functionTestScala extends TestBaseScala with Matchers with GeometrySample
     assert(functionDf.first().get(0) == null)
     functionDf = sparkSession.sql("select ST_Union(null, null)")
     assert(functionDf.first().get(0) == null)
+    functionDf = sparkSession.sql("select ST_PointOnSurface(null)")
+    assert(functionDf.first().get(0) == null)
     functionDf = sparkSession.sql("select ST_Reverse(null)")
     assert(functionDf.first().get(0) == null)
   }