[SEDONA-481] Implements ST_Snap

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

@@ -38,6 +38,7 @@
 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.overlay.snap.GeometrySnapper;
 import org.locationtech.jts.operation.valid.IsSimpleOp;
 import org.locationtech.jts.operation.valid.IsValidOp;
 import org.locationtech.jts.operation.valid.TopologyValidationError;
@@ -1042,6 +1043,11 @@ public static Geometry[] subDivide(Geometry geometry, int maxVertices) {
         return GeometrySubDivider.subDivide(geometry, maxVertices);
     }
 
+    public static Geometry snap(Geometry input, Geometry reference, double tolerance) {
+        GeometrySnapper snapper = new GeometrySnapper(input);
+        return snapper.snapTo(reference, tolerance);
+    }
+
     public static Geometry makeLine(Geometry geom1, Geometry geom2) {
         Geometry[] geoms = new Geometry[]{geom1, geom2};
         return makeLine(geoms);

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

@@ -1187,6 +1187,25 @@ public void testBuffer() {
         assertEquals(expected, actual);
     }
 
+    @Test
+    public void testSnap() throws ParseException {
+        Geometry poly = geomFromWKT("POLYGON ((2.6 12.5, 2.6 20.0, 12.6 20.0, 12.6 12.5, 2.6 12.5 ))", 0);
+        Geometry line = geomFromWKT("LINESTRING (0.5 10.7, 5.4 8.4, 10.1 10.0)", 0);
+        double distance = Functions.distance(poly, line);
+        String actual = Functions.asWKT(Functions.snap(poly, line, distance * 1.01));
+        String expected = "POLYGON ((2.6 12.5, 2.6 20, 12.6 20, 12.6 12.5, 10.1 10, 2.6 12.5))";
+        assertEquals(expected, actual);
+
+        actual = Functions.asWKT(Functions.snap(poly, line, distance * 1.25));
+        expected = "POLYGON ((0.5 10.7, 2.6 20, 12.6 20, 12.6 12.5, 10.1 10, 5.4 8.4, 0.5 10.7))";
+        assertEquals(expected, actual);
+
+        // if the tolerance is less than distance between Geometries then input Geometry will be returned as is.
+        actual = Functions.asWKT(Functions.snap(poly, line, distance * 0.9));
+        expected = Functions.asWKT(poly);
+        assertEquals(expected, actual);
+    }
+
     @Test
     public void testBufferSpheroidal() throws ParseException {
         Geometry polygon1 = GEOMETRY_FACTORY.createPolygon(coordArray(16.2500, 48.2500, 16.3500, 48.2500, 16.3500, 48.2000, 16.2500, 48.2000, 16.2500, 48.2500));

docs/api/flink/Function.md

@@ -2508,6 +2508,38 @@ Output:
 POLYGON ((8 25, 28 22, 15 11, 33 3, 56 30, 47 44, 35 36, 43 19, 24 39, 8 25))
 ```
 
+## ST_Snap
+
+Introduction: Snaps the vertices and segments of the `input` geometry to `reference` geometry within the specified `tolerance` distance. The `tolerance` parameter controls the maximum snap distance.
+
+If the minimum distance between the geometries exceeds the `tolerance`, the `input` geometry is returned unmodified. Adjusting the `tolerance` value allows tuning which vertices should snap to the `reference` and which remain untouched.
+
+Since: `v1.6.0`
+
+Format: `ST_Snap(input: Geometry, reference: Geometry, tolerance: double)`
+
+Input geometry:
+
+<img width="250" src="../../../image/st_snap/st-snap-base-example.png" title="ST_Snap Base example"/>
+
+SQL Example:
+
+```sql
+SELECT ST_Snap(
+        ST_GeomFromWKT('POLYGON ((236877.58 -6.61, 236878.29 -8.35, 236879.98 -8.33, 236879.72 -7.63, 236880.35 -6.62, 236877.58 -6.61), (236878.45 -7.01, 236878.43 -7.52, 236879.29 -7.50, 236878.63 -7.22, 236878.76 -6.89, 236878.45 -7.01))') as poly,
+        ST_GeomFromWKT('LINESTRING (236880.53 -8.22, 236881.15 -7.68, 236880.69 -6.81)') as line,
+       ST_Distance(poly, line) * 1.01
+       )
+```
+
+Output:
+
+<img width="250" src="../../../image/st_snap/st-snap-applied.png" title="ST_Snap applied example"/>
+
+```
+POLYGON ((236877.58 -6.61, 236878.29 -8.35, 236879.98 -8.33, 236879.72 -7.63, 236880.69 -6.81, 236877.58 -6.61), (236878.45 -7.01, 236878.43 -7.52, 236879.29 -7.5, 236878.63 -7.22, 236878.76 -6.89, 236878.45 -7.01))
+```
+
 ## ST_StartPoint
 
 Introduction: Returns first point of given linestring.

docs/api/snowflake/vector-data/Function.md

@@ -1890,6 +1890,36 @@ SELECT ST_SimplifyPreserveTopology(polygondf.countyshape, 10.0)
 FROM polygondf
 ```
 
+## ST_Snap
+
+Introduction: Snaps the vertices and segments of the `input` geometry to `reference` geometry within the specified `tolerance` distance. The `tolerance` parameter controls the maximum snap distance.
+
+If the minimum distance between the geometries exceeds the `tolerance`, the `input` geometry is returned unmodified. Adjusting the `tolerance` value allows tuning which vertices should snap to the `reference` and which remain untouched.
+
+Format: `ST_Snap(input: Geometry, reference: Geometry, tolerance: double)`
+
+Input geometry:
+
+<img width="250" src="../../../../image/st_snap/st-snap-base-example.png" title="ST_Snap Base example"/>
+
+SQL Example:
+
+```sql
+SELECT ST_Snap(
+        ST_GeomFromWKT('POLYGON ((236877.58 -6.61, 236878.29 -8.35, 236879.98 -8.33, 236879.72 -7.63, 236880.35 -6.62, 236877.58 -6.61), (236878.45 -7.01, 236878.43 -7.52, 236879.29 -7.50, 236878.63 -7.22, 236878.76 -6.89, 236878.45 -7.01))') as poly,
+        ST_GeomFromWKT('LINESTRING (236880.53 -8.22, 236881.15 -7.68, 236880.69 -6.81)') as line,
+       ST_Distance(poly, line) * 1.01
+       )
+```
+
+Output:
+
+<img width="250" src="../../../../image/st_snap/st-snap-applied.png" title="ST_Snap applied example"/>
+
+```
+POLYGON ((236877.58 -6.61, 236878.29 -8.35, 236879.98 -8.33, 236879.72 -7.63, 236880.69 -6.81, 236877.58 -6.61), (236878.45 -7.01, 236878.43 -7.52, 236879.29 -7.5, 236878.63 -7.22, 236878.76 -6.89, 236878.45 -7.01))
+```
+
 ## ST_Split
 
 Introduction: Split an input geometry by another geometry (called the blade).

docs/api/sql/Function.md

@@ -2498,6 +2498,38 @@ Output:
 POLYGON ((8 25, 28 22, 15 11, 33 3, 56 30, 47 44, 35 36, 43 19, 24 39, 8 25))
 ```
 
+## ST_Snap
+
+Introduction: Snaps the vertices and segments of the `input` geometry to `reference` geometry within the specified `tolerance` distance. The `tolerance` parameter controls the maximum snap distance.
+
+If the minimum distance between the geometries exceeds the `tolerance`, the `input` geometry is returned unmodified. Adjusting the `tolerance` value allows tuning which vertices should snap to the `reference` and which remain untouched.
+
+Since: `v1.6.0`
+
+Format: `ST_Snap(input: Geometry, reference: Geometry, tolerance: double)`
+
+Input geometry:
+
+<img width="250" src="../../../image/st_snap/st-snap-base-example.png" title="ST_Snap Base example"/>
+
+SQL Example:
+
+```sql
+SELECT ST_Snap(
+        ST_GeomFromWKT('POLYGON ((236877.58 -6.61, 236878.29 -8.35, 236879.98 -8.33, 236879.72 -7.63, 236880.35 -6.62, 236877.58 -6.61), (236878.45 -7.01, 236878.43 -7.52, 236879.29 -7.50, 236878.63 -7.22, 236878.76 -6.89, 236878.45 -7.01))') as poly,
+        ST_GeomFromWKT('LINESTRING (236880.53 -8.22, 236881.15 -7.68, 236880.69 -6.81)') as line,
+       ST_Distance(poly, line) * 1.01
+       )
+```
+
+Output:
+
+<img width="250" src="../../../image/st_snap/st-snap-applied.png" title="ST_Snap applied example"/>
+
+```
+POLYGON ((236877.58 -6.61, 236878.29 -8.35, 236879.98 -8.33, 236879.72 -7.63, 236880.69 -6.81, 236877.58 -6.61), (236878.45 -7.01, 236878.43 -7.52, 236879.29 -7.5, 236878.63 -7.22, 236878.76 -6.89, 236878.45 -7.01))
+```
+
 ## ST_Split
 
 Introduction: Split an input geometry by another geometry (called the blade).

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

@@ -134,6 +134,7 @@ public static UserDefinedFunction[] getFuncs() {
                 new Functions.ST_Subdivide(),
                 new Functions.ST_SymDifference(),
                 new Functions.ST_S2CellIDs(),
+                new Functions.ST_Snap(),
                 new Functions.ST_S2ToGeom(),
                 new Functions.ST_GeometricMedian(),
                 new Functions.ST_NumPoints(),

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

@@ -839,6 +839,17 @@ public Geometry eval(@DataTypeHint(value = "RAW", bridgedTo = org.locationtech.j
         }
     }
 
+    public static class ST_Snap 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,
+                             @DataTypeHint("Double") Double tolerance) {
+            Geometry input = (Geometry) o1;
+            Geometry reference = (Geometry) o2;
+            return org.apache.sedona.common.Functions.snap(input, reference, tolerance);
+        }
+    }
+
     public static class ST_S2CellIDs extends ScalarFunction {
         @DataTypeHint(value = "ARRAY<BIGINT>")
         public Long[] eval(@DataTypeHint(value = "RAW", bridgedTo = org.locationtech.jts.geom.Geometry.class) Object o,

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

@@ -973,6 +973,20 @@ public void testMinimumBoundingRadius() {
         assertEquals(0.5, result.getRight(), 1e-6);
     }
 
+    @Test
+    public void testSnap() {
+        Table base = tableEnv.sqlQuery("SELECT ST_GeomFromWKT('POLYGON((2.6 12.5, 2.6 20.0, 12.6 20.0, 12.6 12.5, 2.6 12.5 ))') AS poly, ST_GeomFromWKT('LINESTRING (0.5 10.7, 5.4 8.4, 10.1 10.0)') AS line");
+        Table table = base.select(call(Functions.ST_Snap.class.getSimpleName(), $("poly"), $("line"), 2.525).as("result"));
+        String actual = (String) first(table.select(call(Functions.ST_AsText.class.getSimpleName(), $("result")))).getField(0);
+        String expected = "POLYGON ((2.6 12.5, 2.6 20, 12.6 20, 12.6 12.5, 10.1 10, 2.6 12.5))";
+        assertEquals(expected, actual);
+
+        table = base.select(call(Functions.ST_Snap.class.getSimpleName(), $("poly"), $("line"), 3.125).as("result"));
+        actual = (String) first(table.select(call(Functions.ST_AsText.class.getSimpleName(), $("result")))).getField(0);
+        expected = "POLYGON ((0.5 10.7, 2.6 20, 12.6 20, 12.6 12.5, 10.1 10, 5.4 8.4, 0.5 10.7))";
+        assertEquals(expected, actual);
+    }
+
     @Test
     public void testMulti() {
         Table table = tableEnv.sqlQuery("SELECT ST_GeomFromWKT('POINT (0 0)') AS geom");

python/sedona/sql/st_functions.py

@@ -1229,6 +1229,17 @@ def ST_SetSRID(geometry: ColumnOrName, srid: Union[ColumnOrName, int]) -> Column
     """
     return _call_st_function("ST_SetSRID", (geometry, srid))
 
+@validate_argument_types
+def ST_Snap(input: ColumnOrName, reference: ColumnOrName, tolerance: Union[ColumnOrName, float]) -> Column:
+    """Snaps input Geometry to reference Geometry controlled by distance tolerance.
+
+    :param input: Geometry
+    :param reference: Geometry to snap to
+    :param tolerance: Distance to control snapping
+    :return: Snapped Geometry
+    """
+    return _call_st_function("ST_Snap", (input, reference, tolerance))
+
 
 @validate_argument_types
 def ST_SRID(geometry: ColumnOrName) -> Column:

python/tests/sql/test_dataframe_api.py

@@ -149,6 +149,7 @@
     (stf.ST_SetSRID, ("point", 3021), "point_geom", "ST_SRID(geom)", 3021),
     (stf.ST_ShiftLongitude, ("geom",), "triangle_geom", "", "POLYGON ((0 0, 1 0, 1 1, 0 0))"),
     (stf.ST_SimplifyPreserveTopology, ("geom", 0.2), "0.9_poly", "", "POLYGON ((0 0, 1 0, 1 1, 0 0))"),
+    (stf.ST_Snap, ("poly", "line", 2.525), "poly_and_line", "" ,"POLYGON ((2.6 12.5, 2.6 20, 12.6 20, 12.6 12.5, 10.1 10, 2.6 12.5))"),
     (stf.ST_Split, ("line", "points"), "multipoint_splitting_line", "", "MULTILINESTRING ((0 0, 0.5 0.5), (0.5 0.5, 1 1), (1 1, 1.5 1.5, 2 2))"),
     (stf.ST_SRID, ("point",), "point_geom", "", 0),
     (stf.ST_StartPoint, ("line",), "linestring_geom", "", "POINT (0 0)"),
@@ -312,6 +313,7 @@
     (stf.ST_ShiftLongitude, (None,)),
     (stf.ST_SimplifyPreserveTopology, (None, 0.2)),
     (stf.ST_SimplifyPreserveTopology, ("", None)),
+    (stf.ST_Snap, (None, None, 12)),
     (stf.ST_SRID, (None,)),
     (stf.ST_StartPoint, (None,)),
     (stf.ST_SubDivide, (None, 5)),
@@ -452,6 +454,8 @@ def base_df(self, request):
             return TestDataFrameAPI.spark.sql("SELECT ST_GeomFromWKT('POINT (-122.335167 47.608013)') AS seattle, ST_GeomFromWKT('POINT (-73.935242 40.730610)') as ny")
         elif request.param == "line_crossing_dateline":
             return TestDataFrameAPI.spark.sql("SELECT ST_GeomFromWKT('LINESTRING (179.95 30, -179.95 30)') AS line")
+        elif request.param == "poly_and_line":
+            return TestDataFrameAPI.spark.sql("SELECT ST_GeomFromWKT('POLYGON((2.6 12.5, 2.6 20.0, 12.6 20.0, 12.6 12.5, 2.6 12.5 ))') as poly, ST_GeomFromWKT('LINESTRING (0.5 10.7, 5.4 8.4, 10.1 10.0)') as line")
         raise ValueError(f"Invalid base_df name passed: {request.param}")
 
     def _id_test_configuration(val):

python/tests/sql/test_function.py

@@ -576,6 +576,17 @@ def test_st_start_point(self):
 
         assert (not polygons.count())
 
+    def test_st_snap(self):
+        baseDf = self.spark.sql("SELECT ST_GeomFromWKT('POLYGON((2.6 12.5, 2.6 20.0, 12.6 20.0, 12.6 12.5, 2.6 12.5 "
+                                "))') AS poly, ST_GeomFromWKT('LINESTRING (0.5 10.7, 5.4 8.4, 10.1 10.0)') AS line")
+        actual = baseDf.selectExpr("ST_AsText(ST_Snap(poly, line, 2.525))").take(1)[0][0]
+        expected = "POLYGON ((2.6 12.5, 2.6 20, 12.6 20, 12.6 12.5, 10.1 10, 2.6 12.5))"
+        assert (expected == actual)
+
+        actual = baseDf.selectExpr("ST_AsText(ST_Snap(poly, line, 3.125))").take(1)[0][0]
+        expected = "POLYGON ((0.5 10.7, 2.6 20, 12.6 20, 12.6 12.5, 10.1 10, 5.4 8.4, 0.5 10.7))"
+        assert (expected == actual)
+
     def test_st_end_point(self):
         linestring_dataframe = create_sample_lines_df(self.spark, 5)
         other_geometry_dataframe = create_sample_points_df(self.spark, 5). \

snowflake-tester/src/test/java/org/apache/sedona/snowflake/snowsql/TestFunctions.java

@@ -801,6 +801,15 @@ public void test_ST_StartPoint() {
                 "POINT (1 2)"
         );
     }
+
+    @Test
+    public void test_ST_Snap() {
+        registerUDF("ST_Snap", byte[].class, byte[].class, double.class);
+        verifySqlSingleRes(
+                "SELECT sedona.ST_AsText(sedona.ST_Snap(ST_GeomFromWKT('POLYGON((2.6 12.5, 2.6 20.0, 12.6 20.0, 12.6 12.5, 2.6 12.5 ))'), ST_GeomFromWKT('LINESTRING (0.5 10.7, 5.4 8.4, 10.1 10.0)'), 2.525))",
+                "POLYGON ((2.6 12.5, 2.6 20, 12.6 20, 12.6 12.5, 10.1 10, 2.6 12.5))"
+        );
+    }
     @Test
     public void test_ST_SubDivide() {
         registerUDF("ST_SubDivide", byte[].class, int.class);

snowflake-tester/src/test/java/org/apache/sedona/snowflake/snowsql/TestFunctionsV2.java

@@ -765,6 +765,14 @@ public void test_ST_StartPoint() {
         );
     }
     @Test
+    public void test_ST_Snap() {
+        registerUDFV2("ST_Snap", String.class, String.class, double.class);
+        verifySqlSingleRes(
+                "SELECT sedona.ST_AsText(sedona.ST_Snap(ST_GeomFromWKT('POLYGON((2.6 12.5, 2.6 20.0, 12.6 20.0, 12.6 12.5, 2.6 12.5 ))'), ST_GeomFromWKT('LINESTRING (0.5 10.7, 5.4 8.4, 10.1 10.0)'), 2.525))",
+                "POLYGON ((2.6 12.5, 2.6 20, 12.6 20, 12.6 12.5, 10.1 10, 2.6 12.5))"
+        );
+    }
+    @Test
     public void test_ST_SubDivide() {
         registerUDFV2("ST_SubDivide", String.class, int.class);
         verifySqlSingleRes(

snowflake/src/main/java/org/apache/sedona/snowflake/snowsql/UDFs.java

@@ -1069,6 +1069,17 @@ public static byte[] ST_StartPoint(byte[] geometry) {
         );
     }
 
+    @UDFAnnotations.ParamMeta(argNames = {"input", "reference", "tolerance"})
+    public static byte[] ST_Snap(byte[] input, byte[] reference, double tolerance) {
+        return GeometrySerde.serialize(
+                Functions.snap(
+                        GeometrySerde.deserialize(input),
+                        GeometrySerde.deserialize(reference),
+                        tolerance
+                )
+        );
+    }
+
     @UDFAnnotations.ParamMeta(argNames = {"geometry", "maxVertices"})
     public static byte[] ST_SubDivide(byte[] geometry, int maxVertices) {
         return GeometrySerde.serialize(

snowflake/src/main/java/org/apache/sedona/snowflake/snowsql/UDFsV2.java

@@ -918,6 +918,17 @@ public static String ST_StartPoint(String geometry) {
         );
     }
 
+    @UDFAnnotations.ParamMeta(argNames = {"input", "reference", "tolerance"}, argTypes = {"Geometry", "Geometry", "double"}, returnTypes = "Geometry")
+    public static String ST_Snap(String input, String reference, double tolerance) {
+        return GeometrySerde.serGeoJson(
+                Functions.snap(
+                        GeometrySerde.deserGeoJson(input),
+                        GeometrySerde.deserGeoJson(reference),
+                        tolerance
+                )
+        );
+    }
+
     @UDFAnnotations.ParamMeta(argNames = {"geometry", "maxVertices"}, argTypes = {"Geometry", "int"}, returnTypes = "Geometry")
     public static String ST_SubDivide(String geometry, int maxVertices) {
         return GeometrySerde.serGeoJson(

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

@@ -107,6 +107,7 @@ object Catalog {
     function[ST_Y](),
     function[ST_Z](),
     function[ST_StartPoint](),
+    function[ST_Snap](),
     function[ST_ClosestPoint](),
     function[ST_Boundary](),
     function[ST_MinimumBoundingRadius](),

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

@@ -445,6 +445,14 @@ case class ST_StartPoint(inputExpressions: Seq[Expression])
   }
 }
 
+case class ST_Snap(inputExpressions: Seq[Expression])
+  extends InferredExpression(Functions.snap _) {
+
+  protected def withNewChildrenInternal(newChildren: IndexedSeq[Expression]) = {
+    copy(inputExpressions = newChildren)
+  }
+}
+
 case class ST_Boundary(inputExpressions: Seq[Expression])
   extends InferredExpression(Functions.boundary _) {
 

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

@@ -287,6 +287,9 @@ object st_functions extends DataFrameAPI {
   def ST_StartPoint(lineString: Column): Column = wrapExpression[ST_StartPoint](lineString)
   def ST_StartPoint(lineString: String): Column = wrapExpression[ST_StartPoint](lineString)
 
+  def ST_Snap(input: Column, reference: Column, tolerance: Column): Column = wrapExpression[ST_Snap](input, reference, tolerance)
+  def ST_Snap(input: String, reference: String, tolerance: Double): Column = wrapExpression[ST_Snap](input, reference, tolerance)
+
   def ST_SubDivide(geometry: Column, maxVertices: Column): Column = wrapExpression[ST_SubDivide](geometry, maxVertices)
   def ST_SubDivide(geometry: String, maxVertices: Int): Column = wrapExpression[ST_SubDivide](geometry, maxVertices)
 

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

@@ -603,6 +603,17 @@ class dataFrameAPITestScala extends TestBaseScala {
       assert(actualResult == expectedResult)
     }
 
+    it("Passed ST_Snap") {
+      val baseDf = sparkSession.sql("SELECT ST_GeomFromWKT('POLYGON((2.6 12.5, 2.6 20.0, 12.6 20.0, 12.6 12.5, 2.6 12.5 ))') AS poly, ST_GeomFromWKT('LINESTRING (0.5 10.7, 5.4 8.4, 10.1 10.0)') AS line")
+      var actual = baseDf.select(ST_Snap("poly", "line", 2.525)).take(1)(0).get(0).asInstanceOf[Geometry].toText
+      var expected = "POLYGON ((2.6 12.5, 2.6 20, 12.6 20, 12.6 12.5, 10.1 10, 2.6 12.5))"
+      assert(expected.equals(actual))
+
+      actual = baseDf.select(ST_Snap("poly", "line", 3.125)).take(1)(0).get(0).asInstanceOf[Geometry].toText
+      expected = "POLYGON ((0.5 10.7, 2.6 20, 12.6 20, 12.6 12.5, 10.1 10, 5.4 8.4, 0.5 10.7))"
+      assert(expected.equals(actual))
+    }
+
     it("Passed ST_Boundary") {
       val baseDf = sparkSession.sql("SELECT ST_GeomFromWKT('POLYGON ((0 0, 1 0, 1 1, 0 0))') AS geom")
       val df = baseDf.select(ST_Boundary("geom"))