[SEDONA-107] Add St_Reverse function

docs/api/flink/Function.md

@@ -99,4 +99,23 @@ Result:
 +-----------------------------+
 |u3r0p                        |
 +-----------------------------+
-```
+```
+
+## ST_Reverse
+
+Introduction: Return the geometry with vertex order reversed
+
+Format: `ST_Reverse (A:geometry)`
+
+Since: `v1.2.1`
+
+Example:
+
+```SQL
+SELECT ST_Reverse(df.geometry) AS geom
+FROM df
+```
+
+Input: `POLYGON ((-0.5 -0.5, -0.5 0.5, 0.5 0.5, 0.5 -0.5, -0.5 -0.5))`
+
+Output: `POLYGON ((-0.5 -0.5, 0.5 -0.5, 0.5 0.5, -0.5 0.5, -0.5 -0.5))`

docs/api/sql/Function.md

@@ -1029,3 +1029,29 @@ Result:
 ```
 POLYGON ((3 -1, 3 -3, -3 -3, -3 3, 3 3, 3 1, 5 0, 3 -1))
 ```
+
+## ST_Reverse
+
+Introduction: Return the geometry with vertex order reversed
+
+Format: `ST_Reverse (A:geometry)`
+
+Since: `v1.2.1`
+
+Example:
+
+```SQL
+SELECT ST_AsText(
+    ST_Reverse(ST_GeomFromText('LINESTRING(0 0, 1 2, 2 4, 3 6)'))
+) AS geom
+```
+
+Result:
+
+```
++---------------------------------------------------------------+
+|geom                                                           |
++---------------------------------------------------------------+
+|LINESTRING (3 6, 2 4, 1 2, 0 0)                                |
++---------------------------------------------------------------+
+```

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

@@ -30,7 +30,8 @@ public static UserDefinedFunction[] getFuncs() {
                 new Functions.ST_Distance(),
                 new Functions.ST_Transform(),
                 new Functions.ST_FlipCoordinates(),
-                new Functions.ST_GeoHash()
+                new Functions.ST_GeoHash(),
+                new Functions.ST_Reverse()
         };
     }
 

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

@@ -84,4 +84,12 @@ public Optional<String> eval(@DataTypeHint(value = "RAW", bridgedTo = org.locati
             return Optional.empty();
         }
     }
+
+    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) {
+            Geometry geom = (Geometry) o;
+            return geom.reverse();
+        }
+    }
 }

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

@@ -64,4 +64,12 @@ public void testGeomToGeoHash() {
         );
         assertEquals(first(pointTable).getField(0), Optional.of("s0000"));
     }
+
+    @Test
+    public void testReverse() {
+        Table polygonTable = createPolygonTable(1);
+        Table ReversedTable = polygonTable.select(call(Functions.ST_Reverse.class.getSimpleName(), $(polygonColNames[0])));
+        Geometry result = (Geometry) first(ReversedTable).getField(0);
+        assertEquals("POLYGON ((-0.5 -0.5, 0.5 -0.5, 0.5 0.5, -0.5 0.5, -0.5 -0.5))", result.toString());
+    }
 }

python/tests/sql/test_function.py

@@ -880,6 +880,39 @@ def test_st_collect_on_multiple_columns(self):
             "MULTIPOLYGON (((1 2, 1 4, 3 4, 3 2, 1 2)), ((0.5 0.5, 5 0, 5 5, 0 5, 0.5 0.5)))"
         })
 
+    def test_st_reverse(self):
+        test_cases = {
+            "'POLYGON((-1 0 0, 1 0 0, 0 0 1, 0 1 0, -1 0 0))'":
+                "POLYGON Z((-1 0 0, 0 1 0, 0 0 1, 1 0 0, -1 0 0))",
+            "'LINESTRING(0 0, 1 2, 2 4, 3 6)'":
+                "LINESTRING (3 6, 2 4, 1 2, 0 0)",
+            "'POINT(1 2)'":
+                "POINT (1 2)",
+            "'MULTIPOINT((10 40 66), (40 30 77), (20 20 88), (30 10 99))'":
+                "MULTIPOINT Z((10 40 66), (40 30 77), (20 20 88), (30 10 99))",
+            "'MULTIPOLYGON(((30 20 11, 45 40 11, 10 40 11, 30 20 11)), " \
+            "((15 5 11, 40 10 11, 10 20 11, 5 10 11, 15 5 11)))'":
+                "MULTIPOLYGON Z(((30 20 11, 10 40 11, 45 40 11, 30 20 11)), " \
+                "((15 5 11, 5 10 11, 10 20 11, 40 10 11, 15 5 11)))",
+            "'MULTILINESTRING((10 10 11, 20 20 11, 10 40 11), " \
+            "(40 40 11, 30 30 11, 40 20 11, 30 10 11))'":
+                "MULTILINESTRING Z((10 40 11, 20 20 11, 10 10 11), " \
+                "(30 10 11, 40 20 11, 30 30 11, 40 40 11))",
+            "'MULTIPOLYGON(((40 40 11, 20 45 11, 45 30 11, 40 40 11)), " \
+            "((20 35 11, 10 30 11, 10 10 11, 30 5 11, 45 20 11, 20 35 11)," \
+            "(30 20 11, 20 15 11, 20 25 11, 30 20 11)))'":
+                "MULTIPOLYGON Z(((40 40 11, 45 30 11, 20 45 11, 40 40 11)), " \
+                "((20 35 11, 45 20 11, 30 5 11, 10 10 11, 10 30 11, 20 35 11), " \
+                "(30 20 11, 20 25 11, 20 15 11, 30 20 11)))",
+            "'POLYGON((0 0 11, 0 5 11, 5 5 11, 5 0 11, 0 0 11), " \
+            "(1 1 11, 2 1 11, 2 2 11, 1 2 11, 1 1 11))'":
+                "POLYGON Z((0 0 11, 5 0 11, 5 5 11, 0 5 11, 0 0 11), " \
+                "(1 1 11, 1 2 11, 2 2 11, 2 1 11, 1 1 11))"
+        }
+        for input_geom, expected_geom in test_cases.items():
+            reversed_geometry = self.spark.sql("select ST_AsText(ST_Reverse(ST_GeomFromText({})))".format(input_geom))
+            assert reversed_geometry.take(1)[0][0] == expected_geom
+
     def calculate_st_is_ring(self, wkt):
         geometry_collected = self.__wkt_list_to_data_frame([wkt]). \
             selectExpr("ST_IsRing(geom) as is_ring") \

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

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

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

@@ -1527,4 +1527,26 @@ case class ST_Multi(inputExpressions: Seq[Expression]) extends UnaryGeometryExpr
   override protected def nullSafeEval(geometry: Geometry): Any ={
     Collect.createMultiGeometry(Seq(geometry)).toGenericArrayData
   }
+}
+
+/**
+ * Returns the geometry with vertex order reversed
+ *
+ * @param inputExpressions
+ */
+case class ST_Reverse(inputExpressions: Seq[Expression])
+  extends UnaryGeometryExpression with CodegenFallback {
+  assert(inputExpressions.length == 1)
+
+  override protected def nullSafeEval(geometry: Geometry): Any = {
+    new GenericArrayData(GeometrySerializer.serialize(geometry.reverse()))
+  }
+
+  override def dataType: DataType = GeometryUDT
+
+  override def children: Seq[Expression] = inputExpressions
+
+  protected def withNewChildrenInternal(newChildren: IndexedSeq[Expression]) = {
+    copy(inputExpressions = newChildren)
+  }
 }

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

@@ -1266,6 +1266,42 @@ class functionTestScala extends TestBaseScala with Matchers with GeometrySample
 
   }
 
+  it ("Should pass ST_Reverse") {
+    val geomTestCases = Map(
+      "'POLYGON((-1 0 0, 1 0 0, 0 0 1, 0 1 0, -1 0 0))'"
+        -> "POLYGON Z((-1 0 0, 0 1 0, 0 0 1, 1 0 0, -1 0 0))",
+      "'LINESTRING(0 0, 1 2, 2 4, 3 6)'"
+        -> "LINESTRING (3 6, 2 4, 1 2, 0 0)",
+      "'POINT(1 2)'"
+        -> "POINT (1 2)",
+      "'MULTIPOINT((10 40 66), (40 30 77), (20 20 88), (30 10 99))'"
+        -> "MULTIPOINT Z((10 40 66), (40 30 77), (20 20 88), (30 10 99))",
+      """'MULTIPOLYGON(((30 20 11, 45 40 11, 10 40 11, 30 20 11)),
+      |((15 5 11, 40 10 11, 10 20 11, 5 10 11, 15 5 11)))'""".stripMargin.replaceAll("\n", " ")
+        -> """MULTIPOLYGON Z(((30 20 11, 10 40 11, 45 40 11, 30 20 11)),
+          |((15 5 11, 5 10 11, 10 20 11, 40 10 11, 15 5 11)))""".stripMargin.replaceAll("\n", " "),
+      """'MULTILINESTRING((10 10 11, 20 20 11, 10 40 11),
+        |(40 40 11, 30 30 11, 40 20 11, 30 10 11))'""".stripMargin.replaceAll("\n", " ")
+        -> """MULTILINESTRING Z((10 40 11, 20 20 11, 10 10 11),
+          |(30 10 11, 40 20 11, 30 30 11, 40 40 11))""".stripMargin.replaceAll("\n", " "),
+      """'MULTIPOLYGON(((40 40 11, 20 45 11, 45 30 11, 40 40 11)),
+      |((20 35 11, 10 30 11, 10 10 11, 30 5 11, 45 20 11, 20 35 11),
+      |(30 20 11, 20 15 11, 20 25 11, 30 20 11)))'""".stripMargin.replaceAll("\n", " ")
+        -> """MULTIPOLYGON Z(((40 40 11, 45 30 11, 20 45 11, 40 40 11)),
+          |((20 35 11, 45 20 11, 30 5 11, 10 10 11, 10 30 11, 20 35 11),
+          |(30 20 11, 20 25 11, 20 15 11, 30 20 11)))""".stripMargin.replaceAll("\n", " "),
+      """'POLYGON((0 0 11, 0 5 11, 5 5 11, 5 0 11, 0 0 11),
+      |(1 1 11, 2 1 11, 2 2 11, 1 2 11, 1 1 11))'""".stripMargin.replaceAll("\n", " ")
+        -> """POLYGON Z((0 0 11, 5 0 11, 5 5 11, 0 5 11, 0 0 11),
+          |(1 1 11, 1 2 11, 2 2 11, 2 1 11, 1 1 11))""".stripMargin.replaceAll("\n", " ")
+    )
+    for((inputGeom, expectedGeom) <- geomTestCases) {
+      var df = sparkSession.sql(s"select ST_AsText(ST_Reverse(ST_GeomFromText($inputGeom)))")
+      var result = df.collect()
+      assert(result.head.get(0).asInstanceOf[String]==expectedGeom)
+    }
+  }
+
   it("handles nulls") {
     var functionDf: DataFrame = null
     functionDf = sparkSession.sql("select ST_Distance(null, null)")
@@ -1372,5 +1408,7 @@ 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_Reverse(null)")
+    assert(functionDf.first().get(0) == null)
   }
 }