[SEDONA-654] Add ST_RotateY

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

@@ -2230,6 +2230,15 @@ public static Geometry rotateX(Geometry geometry, double angle) {
     return affine(geometry, 1, 0, 0, 0, cosAngle, -sinAngle, 0, sinAngle, cosAngle, 0, 0, 0);
   }
 
+  public static Geometry rotateY(Geometry geometry, double angle) {
+    if (GeomUtils.isAnyGeomEmpty(geometry)) {
+      return geometry;
+    }
+    double sinAngle = Math.sin(angle);
+    double cosAngle = Math.cos(angle);
+    return affine(geometry, cosAngle, 0, sinAngle, 0, 1, 0, -sinAngle, 0, cosAngle, 0, 0, 0);
+  }
+
   /**
    * Rotates a geometry by a given angle in radians.
    *

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

@@ -3951,6 +3951,21 @@ public void rotateX() throws ParseException {
     assertEquals(expected, actual);
   }
 
+  @Test
+  public void rotateY() throws ParseException {
+    Geometry lineString = Constructors.geomFromEWKT("LINESTRING (50 160, 50 50, 100 50)");
+    String actual = Functions.asEWKT(Functions.rotateY(lineString, Math.PI));
+    String expected = "LINESTRING (-50 160, -50 50, -100 50)";
+    assertEquals(expected, actual);
+
+    lineString = Constructors.geomFromWKT("LINESTRING(1 2 3, 1 1 1)", 1234);
+    Geometry geomActual = Functions.rotateY(lineString, Math.PI / 2);
+    actual = Functions.asWKT(geomActual);
+    expected = "LINESTRING Z(3 2 -0.9999999999999998, 1 1 -0.9999999999999999)";
+    assertEquals(1234, geomActual.getSRID());
+    assertEquals(expected, actual);
+  }
+
   @Test
   public void rotate() throws ParseException {
     Geometry lineString = Constructors.geomFromEWKT("LINESTRING (50 160, 50 50, 100 50)");

docs/api/flink/Function.md

@@ -3321,6 +3321,26 @@ Output:
 SRID=4326;POLYGON ((0 0, 1 0, 1 -0.8390715290764524, 0 0))
 ```
 
+## ST_RotateY
+
+Introduction: Performs a counter-clockwise rotation of the specified geometry around the Y-axis by the given angle measured in radians.
+
+Format: `ST_RotateY(geometry: Geometry, angle: Double)`
+
+Since: `v1.7.0`
+
+SQL Example:
+
+```sql
+SELECT ST_RotateY(ST_GeomFromEWKT('SRID=4326;POLYGON ((0 0, 1 0, 1 1, 0 0))'), 10)
+```
+
+Output:
+
+```
+SRID=4326;POLYGON ((0 0, -0.8390715290764524 0, -0.8390715290764524 1, 0 0))
+```
+
 ## ST_S2CellIDs
 
 Introduction: Cover the geometry with Google S2 Cells, return the corresponding cell IDs with the given level.

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

@@ -2558,6 +2558,24 @@ Output:
 SRID=4326;POLYGON ((0 0, 1 0, 1 -0.8390715290764524, 0 0))
 ```
 
+## ST_RotateY
+
+Introduction: Performs a counter-clockwise rotation of the specified geometry around the Y-axis by the given angle measured in radians.
+
+Format: `ST_RotateY(geometry: Geometry, angle: Double)`
+
+SQL Example:
+
+```sql
+SELECT ST_RotateY(ST_GeomFromEWKT('SRID=4326;POLYGON ((0 0, 1 0, 1 1, 0 0))'), 10)
+```
+
+Output:
+
+```
+SRID=4326;POLYGON ((0 0, -0.8390715290764524 0, -0.8390715290764524 1, 0 0))
+```
+
 ## ST_S2CellIDs
 
 Introduction: Cover the geometry with Google S2 Cells, return the corresponding cell IDs with the given level.

docs/api/sql/Function.md

@@ -3401,6 +3401,26 @@ Output:
 SRID=4326;POLYGON ((0 0, 1 0, 1 -0.8390715290764524, 0 0))
 ```
 
+## ST_RotateY
+
+Introduction: Performs a counter-clockwise rotation of the specified geometry around the Y-axis by the given angle measured in radians.
+
+Format: `ST_RotateY(geometry: Geometry, angle: Double)`
+
+Since: `v1.7.0`
+
+SQL Example:
+
+```sql
+SELECT ST_RotateY(ST_GeomFromEWKT('SRID=4326;POLYGON ((0 0, 1 0, 1 1, 0 0))'), 10)
+```
+
+Output:
+
+```
+SRID=4326;POLYGON ((0 0, -0.8390715290764524 0, -0.8390715290764524 1, 0 0))
+```
+
 ## ST_S2CellIDs
 
 Introduction: Cover the geometry with Google S2 Cells, return the corresponding cell IDs with the given level.

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

@@ -104,6 +104,7 @@ public static UserDefinedFunction[] getFuncs() {
       new Functions.ST_Reverse(),
       new Functions.ST_Rotate(),
       new Functions.ST_RotateX(),
+      new Functions.ST_RotateY(),
       new Functions.ST_GeometryN(),
       new Functions.ST_InteriorRingN(),
       new Functions.ST_PointN(),

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

@@ -1954,6 +1954,16 @@ public Geometry eval(
     }
   }
 
+  public static class ST_RotateY 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,
+        @DataTypeHint(value = "Double") Double angle) {
+      Geometry geom = (Geometry) o;
+      return org.apache.sedona.common.Functions.rotateY(geom, angle);
+    }
+  }
+
   public static class ST_Rotate extends ScalarFunction {
     @DataTypeHint(value = "RAW", bridgedTo = org.locationtech.jts.geom.Geometry.class)
     public Geometry eval(

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

@@ -2644,6 +2644,22 @@ public void testRotateX() {
     assertEquals(expected, actual);
   }
 
+  @Test
+  public void testRotateY() {
+    Table tbl =
+        tableEnv.sqlQuery(
+            "SELECT ST_GeomFromEWKT('POLYGON ((0 0, 2 0, 1 1, 2 2, 0 2, 1 1, 0 0))') AS geom");
+    String actual =
+        (String)
+            first(
+                    tbl.select(call(Functions.ST_RotateY.class.getSimpleName(), $("geom"), Math.PI))
+                        .as("geom")
+                        .select(call(Functions.ST_AsEWKT.class.getSimpleName(), $("geom"))))
+                .getField(0);
+    String expected = "POLYGON ((0 0, -2 0, -1 1, -2 2, 0 2, -1 1, 0 0))";
+    assertEquals(expected, actual);
+  }
+
   @Test
   public void testRotate() {
     Table tbl =

python/sedona/sql/st_functions.py

@@ -2050,6 +2050,20 @@ def ST_RotateX(geometry: ColumnOrName, angle: Union[ColumnOrName, float]) -> Col
     return _call_st_function("ST_RotateX", (geometry, angle))
 
 
+@validate_argument_types
+def ST_RotateY(geometry: ColumnOrName, angle: Union[ColumnOrName, float]) -> Column:
+    """Returns geometry rotated by the given angle in Y axis
+
+    @param geometry: Geometry column or name
+    :type geometry: ColumnOrName
+    @param angle: Rotation angle in radians
+    :type angle: float
+    @return: Y-axis rotated geometry
+    """
+
+    return _call_st_function("ST_RotateY", (geometry, angle))
+
+
 @validate_argument_types
 def ST_Rotate(geometry: ColumnOrName, angle: Union[ColumnOrName, float], originX: Union[ColumnOrName, float] = None,
               originY: Union[ColumnOrName, float] = None, pointOrigin: ColumnOrName = None) -> Column:

python/tests/sql/test_dataframe_api.py

@@ -210,6 +210,7 @@
     (stf.ST_RemoveRepeatedPoints, ("geom",), "repeated_multipoint", "", "MULTIPOINT (1 1, 2 2, 3 3, 4 4)"),
     (stf.ST_Reverse, ("line",), "linestring_geom", "", "LINESTRING (5 0, 4 0, 3 0, 2 0, 1 0, 0 0)"),
     (stf.ST_RotateX, ("line", 10.0), "4D_line", "ST_ReducePrecision(geom, 2)", "LINESTRING Z (1 -0.3 -1.383092639965822, 2 -0.59 -2.766185279931644, 3 -0.89 -4.149277919897466, -1 0.3 1.383092639965822)"),
+    (stf.ST_RotateY, ("line", 10.0), "4D_line", "ST_AsText(ST_ReducePrecision(geom, 2))", "LINESTRING ZM(-1.38 1 -0.2950504181870827 1, -2.77 2 -0.5901008363741653 2, -4.15 3 -0.8851512545612479 3, 1.38 -1 0.2950504181870827 -1)"),
     (stf.ST_Rotate, ("line", 10.0), "linestring_geom", "ST_ReducePrecision(geom, 2)", "LINESTRING (0 0, -0.84 -0.54, -1.68 -1.09, -2.52 -1.63, -3.36 -2.18, -4.2 -2.72)"),
     (stf.ST_Rotate, ("line", 10.0, 0.0, 0.0), "linestring_geom", "ST_ReducePrecision(geom, 2)", "LINESTRING (0 0, -0.84 -0.54, -1.68 -1.09, -2.52 -1.63, -3.36 -2.18, -4.2 -2.72)"),
     (stf.ST_S2CellIDs, ("point", 30), "point_geom", "", [1153451514845492609]),

python/tests/sql/test_function.py

@@ -960,6 +960,17 @@ def test_st_rotate_x(self):
         expected = "LINESTRING Z (1 -3 2, 1 -0.9999999999999999 1)"
         assert expected == actual
 
+    def test_st_rotate_y(self):
+        baseDf = self.spark.sql("SELECT ST_GeomFromWKT('LINESTRING (50 160, 50 50, 100 50)') as geom1, ST_GeomFromWKT('LINESTRING(1 2 3, 1 1 1)') AS geom2")
+
+        actual = baseDf.selectExpr("ST_RotateY(geom1, PI())").first()[0].wkt
+        expected = "LINESTRING (-50 160, -50 50, -100 50)"
+        assert expected == actual
+
+        actual = baseDf.selectExpr("ST_RotateY(geom2, PI() / 2)").first()[0].wkt
+        expected = "LINESTRING Z (3 2 -0.9999999999999998, 1 1 -0.9999999999999999)"
+        assert expected == actual
+
     def test_st_remove_point(self):
         result_and_expected = [
             [self.calculate_st_remove("Linestring(0 0, 1 1, 1 0, 0 0)", 0), "LINESTRING (1 1, 1 0, 0 0)"],

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

@@ -1227,6 +1227,14 @@ public void test_ST_RotateX() {
         "LINESTRING (0 0, 1 0, 1 -0.8390715290764524, 0 0)");
   }
 
+  @Test
+  public void test_ST_RotateY() {
+    registerUDF("ST_RotateY", byte[].class, double.class);
+    verifySqlSingleRes(
+        "SELECT sedona.ST_AsText(sedona.ST_RotateY(sedona.ST_GeomFromWKT('LINESTRING (0 0, 1 0, 1 1, 0 0)'), 10))",
+        "LINESTRING (0 0, -0.8390715290764524 0, -0.8390715290764524 1, 0 0)");
+  }
+
   @Test
   public void test_ST_Rotate() {
     registerUDF("ST_Rotate", byte[].class, double.class);

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

@@ -1183,6 +1183,15 @@ public void test_ST_RotateX() {
         "LINESTRING(0 0,1 0,1 -0.84,0 0)");
   }
 
+  @Test
+  public void test_ST_RotateY() {
+    registerUDFV2("ST_RotateY", String.class, double.class);
+    registerUDFV2("ST_ReducePrecision", String.class, int.class);
+    verifySqlSingleRes(
+        "select ST_AsText(ST_ReducePrecision(sedona.ST_RotateY(ST_GeometryFromWKT('LINESTRING (0 0, 1 0, 1 1, 0 0)'), 10),2))",
+        "LINESTRING(0 0,-0.84 0,-0.84 1,0 0)");
+  }
+
   @Test
   public void test_ST_Rotate() {
     registerUDFV2("ST_Rotate", String.class, double.class);

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

@@ -1277,6 +1277,11 @@ public static byte[] ST_RotateX(byte[] geometry, double angle) {
     return GeometrySerde.serialize(Functions.rotateX(GeometrySerde.deserialize(geometry), angle));
   }
 
+  @UDFAnnotations.ParamMeta(argNames = {"geometry", "angle"})
+  public static byte[] ST_RotateY(byte[] geometry, double angle) {
+    return GeometrySerde.serialize(Functions.rotateY(GeometrySerde.deserialize(geometry), angle));
+  }
+
   @UDFAnnotations.ParamMeta(argNames = {"geom", "angle"})
   public static byte[] ST_Rotate(byte[] geom, double angle) {
     return GeometrySerde.serialize(Functions.rotate(GeometrySerde.deserialize(geom), angle));

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

@@ -1516,6 +1516,14 @@ public static String ST_RotateX(String geometry, double angle) {
     return GeometrySerde.serGeoJson(Functions.rotateX(GeometrySerde.deserGeoJson(geometry), angle));
   }
 
+  @UDFAnnotations.ParamMeta(
+      argNames = {"geometry", "angle"},
+      argTypes = {"Geometry", "double"},
+      returnTypes = "Geometry")
+  public static String ST_RotateY(String geometry, double angle) {
+    return GeometrySerde.serGeoJson(Functions.rotateY(GeometrySerde.deserGeoJson(geometry), angle));
+  }
+
   @UDFAnnotations.ParamMeta(
       argNames = {"geom", "angle"},
       argTypes = {"Geometry", "double"},

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

@@ -234,6 +234,7 @@ object Catalog {
     function[ST_IsValidReason](),
     function[ST_Rotate](),
     function[ST_RotateX](),
+    function[ST_RotateY](),
     // Expression for rasters
     function[RS_NormalizedDifference](),
     function[RS_Mean](),

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

@@ -1715,6 +1715,13 @@ case class ST_RotateX(inputExpressions: Seq[Expression])
     copy(inputExpressions = newChildren)
 }
 
+case class ST_RotateY(inputExpressions: Seq[Expression])
+    extends InferredExpression(inferrableFunction2(Functions.rotateY)) {
+
+  protected def withNewChildrenInternal(newChildren: IndexedSeq[Expression]) =
+    copy(inputExpressions = newChildren)
+}
+
 case class ST_Rotate(inputExpressions: Seq[Expression])
     extends InferredExpression(
       inferrableFunction2(Functions.rotate),

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

@@ -490,6 +490,13 @@ object st_functions extends DataFrameAPI {
   def ST_RotateX(geometry: String, angle: String): Column =
     wrapExpression[ST_RotateX](geometry, angle)
 
+  def ST_RotateY(geometry: Column, angle: Column): Column =
+    wrapExpression[ST_RotateY](geometry, angle)
+  def ST_RotateY(geometry: String, angle: Double): Column =
+    wrapExpression[ST_RotateY](geometry, angle)
+  def ST_RotateY(geometry: String, angle: String): Column =
+    wrapExpression[ST_RotateY](geometry, angle)
+
   def ST_Rotate(geometry: Column, angle: Column): Column =
     wrapExpression[ST_Rotate](geometry, angle)
   def ST_Rotate(geometry: String, angle: Double): Column =

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

@@ -2282,6 +2282,26 @@ class dataFrameAPITestScala extends TestBaseScala {
       }
     }
 
+    it("Should pass ST_RotateY") {
+      val geomTestCases = Map(
+        (
+          1,
+          "'LINESTRING (50 160, 50 50, 100 50)'",
+          Math.PI) -> "'LINESTRING (-50 160, -50 50, -100 50)'",
+        (
+          2,
+          "'LINESTRING(1 2 3, 1 1 1)'",
+          Math.PI / 2) -> "LINESTRING Z(3 2 -0.9999999999999998, 1 1 -0.9999999999999999)")
+
+      for (((index, geom, angle), expectedResult) <- geomTestCases) {
+        val baseDf = sparkSession.sql(s"SELECT ST_GeomFromEWKT($geom) as geom")
+        val df = baseDf.select(ST_AsEWKT(ST_RotateY("geom", angle)))
+
+        val actual = df.take(1)(0).get(0).asInstanceOf[String]
+        assert(actual == expectedResult.stripPrefix("'").stripSuffix("'"))
+      }
+    }
+
     it("Passed ST_Rotate") {
       val baseDf = sparkSession.sql(
         "SELECT ST_GeomFromEWKT('SRID=4326;POLYGON ((0 0, 2 0, 2 2, 0 2, 1 1, 0 0))') AS geom1, ST_GeomFromText('POINT (2 2)') AS geom2")

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

@@ -3459,6 +3459,33 @@ class functionTestScala
     }
   }
 
+  it("Should pass ST_RotateY") {
+    val geomTestCases = Map(
+      (
+        1,
+        "'LINESTRING (50 160, 50 50, 100 50)'",
+        "PI()") -> "'LINESTRING (-50 160, -50 50, -100 50)'",
+      (
+        2,
+        "'LINESTRING(1 2 3, 1 1 1)'",
+        "PI()/2") -> "LINESTRING Z(3 2 -0.9999999999999998, 1 1 -0.9999999999999999)")
+
+    for (((index, geom, angle), expectedResult) <- geomTestCases) {
+      val df = sparkSession.sql(s"""
+                                   |SELECT
+                                   |  ST_AsEWKT(
+                                   |    ST_RotateY(
+                                   |      ST_GeomFromEWKT($geom),
+                                   |      $angle
+                                   |    )
+                                   |  ) AS geom
+         """.stripMargin)
+
+      val actual = df.take(1)(0).get(0).asInstanceOf[String]
+      assert(actual == expectedResult.stripPrefix("'").stripSuffix("'"))
+    }
+  }
+
   it("Should pass ST_Rotate") {
     val geomTestCases = Map(
       (