[SEDONA-298] Implement ST_ClosestPoint (#877)

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

@@ -30,6 +30,7 @@
 import org.locationtech.jts.io.gml2.GMLWriter;
 import org.locationtech.jts.io.kml.KMLWriter;
 import org.locationtech.jts.linearref.LengthIndexedLine;
+import org.locationtech.jts.operation.distance.DistanceOp;
 import org.locationtech.jts.operation.distance3d.Distance3DOp;
 import org.locationtech.jts.operation.linemerge.LineMerger;
 import org.locationtech.jts.operation.valid.IsSimpleOp;
@@ -447,6 +448,17 @@ public static Geometry lineFromMultiPoint(Geometry geometry) {
         return GEOMETRY_FACTORY.createLineString(coordinates.toArray(new Coordinate[0]));
     }
 
+    public static Geometry closestPoint(Geometry left, Geometry right) {
+        DistanceOp distanceOp = new DistanceOp(left, right);
+        try {
+            Coordinate[] closestPoints = distanceOp.nearestPoints();
+            return GEOMETRY_FACTORY.createPoint(closestPoints[0]);
+        }
+        catch (Exception e) {
+            throw new IllegalArgumentException("ST_ClosestPoint doesn't support empty geometry object.");
+        }
+    }
+
     public static Geometry concaveHull(Geometry geometry, double pctConvex, boolean allowHoles){
         ConcaveHull concave_hull = new ConcaveHull(geometry);
         concave_hull.setMaximumEdgeLengthRatio(pctConvex);

common/src/test/java/org/apache/sedona/common/FunctionsTest.java

@@ -30,7 +30,6 @@
 import java.util.stream.Collectors;
 
 import static org.junit.Assert.*;
-import static org.junit.Assert.assertEquals;
 
 public class FunctionsTest {
     public static final GeometryFactory GEOMETRY_FACTORY = new GeometryFactory();
@@ -310,7 +309,7 @@ public void dimensionGeometry3D() {
         Integer expectedResult = 0;
         assertEquals(actualResult, expectedResult);
 
-        LineString lineString3D = GEOMETRY_FACTORY.createLineString(coordArray(1, 0, 1, 1, 1, 2));
+        LineString lineString3D = GEOMETRY_FACTORY.createLineString(coordArray3d(1, 0, 1, 1, 1, 2));
         actualResult = Functions.dimension(lineString3D);
         expectedResult = 1;
         assertEquals(actualResult, expectedResult);
@@ -1307,6 +1306,75 @@ public void geometryTypeWithMeasuredCollection() {
         assertEquals(expected4, actual4);
     }
 
+    @Test
+    public void closestPoint() {
+        Point point1 = GEOMETRY_FACTORY.createPoint(new Coordinate(1, 1));
+        LineString lineString1 = GEOMETRY_FACTORY.createLineString(coordArray(1, 0, 1, 1, 2, 1, 2, 0, 1, 0));
+        String expected1 = "POINT (1 1)";
+        String actual1 = Functions.closestPoint(point1, lineString1).toText();
+        assertEquals(expected1, actual1);
+
+        Point point2 = GEOMETRY_FACTORY.createPoint(new Coordinate(160, 40));
+        LineString lineString2 = GEOMETRY_FACTORY.createLineString(coordArray(10, 30, 50, 50, 30, 110, 70, 90, 180, 140, 130, 190));
+        String expected2 = "POINT (160 40)";
+        String actual2 = Functions.closestPoint(point2, lineString2).toText();
+        assertEquals(expected2, actual2);
+        Point expectedPoint3 = GEOMETRY_FACTORY.createPoint(new Coordinate(125.75342465753425, 115.34246575342466));
+        Double expected3 = Functions.closestPoint(lineString2, point2).distance(expectedPoint3);
+        assertEquals(expected3, 0, 1e-6);
+
+        Point point4 = GEOMETRY_FACTORY.createPoint(new Coordinate(80, 160));
+        Polygon polygonA = GEOMETRY_FACTORY.createPolygon(coordArray(190, 150, 20, 10, 160, 70, 190, 150));
+        Geometry polygonB = Functions.buffer(point4, 30);
+        Point expectedPoint4 = GEOMETRY_FACTORY.createPoint(new Coordinate(131.59149149528952, 101.89887534906197));
+        Double expected4 = Functions.closestPoint(polygonA, polygonB).distance(expectedPoint4);
+        assertEquals(expected4, 0, 1e-6);
+    }
+
+    @Test
+    public void closestPoint3d() {
+        // One of the object is 3D
+        Point point1 = GEOMETRY_FACTORY.createPoint(new Coordinate(1, 1,10000));
+        LineString lineString1 = GEOMETRY_FACTORY.createLineString(coordArray(1, 0, 1, 1, 2, 1, 2, 0, 1, 0));
+        String expected1 = "POINT (1 1)";
+        String actual1 = Functions.closestPoint(point1, lineString1).toText();
+        assertEquals(expected1, actual1);
+
+        // Both of the object are 3D
+        LineString lineString3D = GEOMETRY_FACTORY.createLineString(coordArray3d(1, 0, 100, 1, 1, 20, 2, 1, 40, 2, 0, 60, 1, 0, 70));
+        String expected2 = "POINT (1 1)";
+        String actual2 = Functions.closestPoint(point1, lineString3D).toText();
+        assertEquals(expected2, actual2);
+    }
+
+    @Test
+    public void closestPointGeomtryCollection() {
+        LineString line = GEOMETRY_FACTORY.createLineString(coordArray(2, 0, 0, 2));
+        Geometry[] geometry = new Geometry[] {
+                GEOMETRY_FACTORY.createLineString(coordArray(2, 0, 2, 1)),
+                GEOMETRY_FACTORY.createPolygon(coordArray(0.0, 0.0, 1.0, 1.0, 1.0, 0.0, 0.0, 0.0))
+        };
+        GeometryCollection geometryCollection = GEOMETRY_FACTORY.createGeometryCollection(geometry);
+        String actual1 = Functions.closestPoint(line, geometryCollection).toText();
+        String expected1 = "POINT (2 0)";
+        assertEquals(actual1 ,expected1);
+    }
+
+    @Test
+    public void closestPointEmpty() {
+        // One of the object is empty
+        Point point = GEOMETRY_FACTORY.createPoint(new Coordinate(1, 1));
+        LineString emptyLineString = GEOMETRY_FACTORY.createLineString();
+        String expected = "ST_ClosestPoint doesn't support empty geometry object.";
+        Exception e1 = assertThrows(IllegalArgumentException.class, () -> Functions.closestPoint(point, emptyLineString));
+        assertEquals(expected, e1.getMessage());
+
+        // Both objects are empty
+        Polygon emptyPolygon = GEOMETRY_FACTORY.createPolygon();
+        Exception e2 = assertThrows(IllegalArgumentException.class, () -> Functions.closestPoint(emptyPolygon, emptyLineString));
+        assertEquals(expected, e2.getMessage());
+    }
+
     @Test
     public void hausdorffDistanceDefaultGeom2D() throws Exception {
         Polygon polygon1 = GEOMETRY_FACTORY.createPolygon(coordArray3d(1, 0, 1, 1, 1, 2, 2, 1, 5, 2, 0, 1, 1, 0, 1));

docs/api/flink/Function.md

@@ -445,6 +445,32 @@ SELECT ST_Centroid(polygondf.countyshape)
 FROM polygondf
 ```
 
+## ST_ClosestPoint
+
+Introduction: Returns the 2-dimensional point on geom1 that is closest to geom2. This is the first point of the shortest line between the geometries. If using 3D geometries, the Z coordinates will be ignored. If you have a 3D Geometry, you may prefer to use ST_3DClosestPoint.
+It will throw an exception indicates illegal argument if one of the params is an empty geometry.
+
+Format: `ST_ClosestPoint(g1: geomtry, g2: geometry)`
+
+Since: `1.5.0`
+
+Example1:
+```sql
+SELECT ST_AsText( ST_ClosestPoint(g1, g2)) As ptwkt;
+```
+
+Input: `g1: POINT (160 40), g2: LINESTRING (10 30, 50 50, 30 110, 70 90, 180 140, 130 190)`
+
+Output: `POINT(160 40)`
+
+Input: `g1: LINESTRING (10 30, 50 50, 30 110, 70 90, 180 140, 130 190), g2: POINT (160 40)`
+
+Output: `POINT(125.75342465753425 115.34246575342466)`
+
+Input: `g1: 'POLYGON ((190 150, 20 10, 160 70, 190 150))', g2: ST_Buffer('POINT(80 160)', 30)`
+
+Output: `POINT(131.59149149528952 101.89887534906197)`
+
 ## ST_CollectionExtract
 
 Introduction: Returns a homogeneous multi-geometry from a given geometry collection.

docs/api/sql/Function.md

@@ -445,6 +445,32 @@ SELECT ST_Centroid(polygondf.countyshape)
 FROM polygondf
 ```
 
+## ST_ClosestPoint
+
+Introduction: Returns the 2-dimensional point on geom1 that is closest to geom2. This is the first point of the shortest line between the geometries. If using 3D geometries, the Z coordinates will be ignored. If you have a 3D Geometry, you may prefer to use ST_3DClosestPoint.
+It will throw an exception indicates illegal argument if one of the params is an empty geometry.
+
+Format: `ST_ClosestPoint(g1: geomtry, g2: geometry)`
+
+Since: `1.5.0`
+
+Example1:
+```sql
+SELECT ST_AsText( ST_ClosestPoint(g1, g2)) As ptwkt;
+```
+
+Input: `g1: POINT (160 40), g2: LINESTRING (10 30, 50 50, 30 110, 70 90, 180 140, 130 190)`
+
+Output: `POINT(160 40)`
+
+Input: `g1: LINESTRING (10 30, 50 50, 30 110, 70 90, 180 140, 130 190), g2: POINT (160 40)`
+
+Output: `POINT(125.75342465753425 115.34246575342466)`
+
+Input: `g1: 'POLYGON ((190 150, 20 10, 160 70, 190 150))', g2: ST_Buffer('POINT(80 160)', 30)`
+
+Output: `POINT(131.59149149528952 101.89887534906197)`
+
 ## ST_Collect
 
 Introduction: Returns MultiGeometry object based on geometry column/s or array with geometries

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

@@ -44,6 +44,7 @@ public static UserDefinedFunction[] getFuncs() {
                 new Functions.ST_Azimuth(),
                 new Functions.ST_Boundary(),
                 new Functions.ST_Buffer(),
+                new Functions.ST_ClosestPoint(),
                 new Functions.ST_Centroid(),
                 new Functions.ST_CollectionExtract(),
                 new Functions.ST_ConcaveHull(),

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

@@ -71,6 +71,16 @@ public Geometry eval(@DataTypeHint(value = "RAW", bridgedTo = org.locationtech.j
         }
     }
 
+    public static class ST_ClosestPoint 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 g1,
+        @DataTypeHint(value = "RAW", bridgedTo = org.locationtech.jts.geom.Geometry.class) Object g2) {
+            Geometry geom1 = (Geometry) g1;
+            Geometry geom2 = (Geometry) g2;
+            return org.apache.sedona.common.Functions.closestPoint(geom1, geom2);
+        }
+    }
+
     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) {

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

@@ -85,6 +85,12 @@ public void testBuffer() {
     }
 
     @Test
+    public void testClosestPoint() {
+        Table table = tableEnv.sqlQuery("SELECT ST_GeomFromWKT('POINT (160 40)') AS g1, ST_GeomFromWKT('POINT (10 10)') as g2");
+        table = table.select(call(Functions.ST_ClosestPoint.class.getSimpleName(), $("g1"), $("g2")));
+        Geometry result = (Geometry) first(table).getField(0);
+        assertEquals("POINT (160 40)", result.toString());
+    }   
     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")));

python/sedona/sql/st_functions.py

@@ -43,6 +43,7 @@
     "ST_Centroid",
     "ST_Collect",
     "ST_CollectionExtract",
+    "ST_ClosestPoint",
     "ST_ConcaveHull",
     "ST_ConvexHull",
     "ST_Difference",
@@ -375,6 +376,21 @@ def ST_CollectionExtract(collection: ColumnOrName, geom_type: Optional[Union[Col
     return _call_st_function("ST_CollectionExtract", args)
 
 
+@validate_argument_types
+def ST_ClosestPoint(a: ColumnOrName, b: ColumnOrName) -> Column:
+    """Returns the 2-dimensional point on geom1 that is closest to geom2. 
+    This is the first point of the shortest line between the geometries.
+
+    :param a: Geometry column to use in the calculation.
+    :type a: ColumnOrName
+    :param b: Geometry column to use in the calculation.
+    :type b: ColumnOrName
+    :return: the 2-dimensional point on a that is closest to b.
+    :rtype: Column
+    """
+    return _call_st_function("ST_ClosestPoint", (a, b))
+
+
 @validate_argument_types
 def ST_ConcaveHull(geometry: ColumnOrName, pctConvex: Union[ColumnOrName, float], allowHoles: Optional[Union[ColumnOrName, bool]] = None) -> Column:
     """Generate the cancave hull of a geometry column.

python/tests/sql/test_dataframe_api.py

@@ -76,6 +76,7 @@
     (stf.ST_Centroid, ("geom",), "triangle_geom", "ST_PrecisionReduce(geom, 2)", "POINT (0.67 0.33)"),
     (stf.ST_Collect, (lambda: f.expr("array(a, b)"),), "two_points", "", "MULTIPOINT Z (0 0 0, 3 0 4)"),
     (stf.ST_Collect, ("a", "b"), "two_points", "", "MULTIPOINT Z (0 0 0, 3 0 4)"),
+    (stf.ST_ClosestPoint, ("point", "line",), "point_and_line", "", "POINT (0 1)"),
     (stf.ST_CollectionExtract, ("geom",), "geom_collection", "", "MULTILINESTRING ((0 0, 1 0))"),
     (stf.ST_CollectionExtract, ("geom", 1), "geom_collection", "", "MULTIPOINT (0 0)"),
     (stf.ST_ConcaveHull, ("geom", 1.0), "triangle_geom", "", "POLYGON ((0 0, 1 1, 1 0, 0 0))"),

python/tests/sql/test_function.py

@@ -899,6 +899,13 @@ def test_geom_from_geohash_precision(self):
         for wkt, expected_wkt in geohash:
             assert wkt == expected_wkt
 
+    def test_st_closest_point(self):
+        expected = "POINT (0 1)"
+        actual_df = self.spark.sql("select ST_AsText(ST_ClosestPoint(ST_GeomFromText('POINT (0 1)'), "
+                                   "ST_GeomFromText('LINESTRING (0 0, 1 0, 2 0, 3 0, 4 0, 5 0)')))")
+        actual = actual_df.take(1)[0][0]
+        assert expected == actual
+
     def test_st_collect_on_array_type(self):
         # given
         geometry_df = self.spark.createDataFrame([

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

@@ -99,6 +99,7 @@ object Catalog {
     function[ST_Y](),
     function[ST_Z](),
     function[ST_StartPoint](),
+    function[ST_ClosestPoint](),
     function[ST_Boundary](),
     function[ST_MinimumBoundingRadius](),
     function[ST_MinimumBoundingCircle](BufferParameters.DEFAULT_QUADRANT_SEGMENTS),

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

@@ -601,6 +601,14 @@ case class ST_SetPoint(inputExpressions: Seq[Expression])
   }
 }
 
+case class ST_ClosestPoint(inputExpressions: Seq[Expression])
+  extends InferredExpression(Functions.closestPoint _) {
+
+  protected def withNewChildrenInternal(newChildren: IndexedSeq[Expression]) = {
+    copy(inputExpressions = newChildren)
+  }
+}
+
 case class ST_IsRing(inputExpressions: Seq[Expression])
   extends InferredExpression(ST_IsRing.isRing _) {
 

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

@@ -73,6 +73,9 @@ object st_functions extends DataFrameAPI {
   def ST_Centroid(geometry: Column): Column = wrapExpression[ST_Centroid](geometry)
   def ST_Centroid(geometry: String): Column = wrapExpression[ST_Centroid](geometry)
 
+  def ST_ClosestPoint(a: Column, b: Column): Column = wrapExpression[ST_ClosestPoint](a, b)
+  def ST_ClosestPoint(a: String, b: String): Column = wrapExpression[ST_ClosestPoint](a, b)
+
   def ST_Collect(geoms: Column): Column = wrapExpression[ST_Collect](geoms)
   def ST_Collect(geoms: String): Column = wrapExpression[ST_Collect](geoms)
   def ST_Collect(geoms: Any*): Column = wrapVarArgExpression[ST_Collect](geoms)

sql/common/src/test/scala/org/apache/sedona/sql/dataFrameAPITestScala.scala

@@ -702,6 +702,14 @@ class dataFrameAPITestScala extends TestBaseScala {
       assert(actualResult == expectedResult)
     }
 
+    it("Passed ST_ClosestPoint") {
+      val polyDf = sparkSession.sql("SELECT ST_GeomFromWKT('POINT (0 1)') as g1, ST_GeomFromWKT('LINESTRING (0 0, 1 0, 2 0, 3 0, 4 0, 5 0)') as g2")
+      val df = polyDf.select(ST_ClosestPoint("g1", "g2"))
+      val expected = "POINT (0 1)"
+      val actual = df.take(1)(0).get(0).asInstanceOf[Geometry].toText()
+      assertEquals(expected, actual)
+    }
+
     it ("Passed ST_AsEWKT") {
       val baseDf = sparkSession.sql("SELECT ST_SetSRID(ST_Point(0.0, 0.0), 4326) AS point")
       val df = baseDf.select(ST_AsEWKT("point"))

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

@@ -1913,6 +1913,22 @@ class functionTestScala extends TestBaseScala with Matchers with GeometrySample
     }
   }
 
+  it("should pass ST_ClosestPoint") {
+    val geomTestCases = Map(
+      ("'POINT (160 40)'", "'LINESTRING (10 30, 50 50, 30 110, 70 90, 180 140, 130 190)'") -> "POINT (160 40)",
+      ("'LINESTRING (0 0, 100 0)'", "'LINESTRING (0 0, 50 50, 100 0)'") -> "POINT (0 0)"
+    )
+    for (((geom), expectedResult) <- geomTestCases) {
+      val g1 = geom._1
+      val g2 = geom._2
+      val df = sparkSession.sql(s"SELECT ST_ClosestPoint(ST_GeomFromWKT($g1), ST_GeomFromWKT($g2))")
+      val actual = df.take(1)(0).get(0).asInstanceOf[Geometry].toText
+      val expected = expectedResult
+      assertEquals(expected, actual)
+
+    }
+  }
+
   it("Should pass ST_AreaSpheroid") {
     val geomTestCases = Map(
       ("'POINT (51.3168 -0.56)'") -> "0.0",