feat: add ST_Scale and ST_ScaleGeom

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

@@ -2266,6 +2266,45 @@ public static Geometry points(Geometry geometry) {
     return geometry.getFactory().createMultiPointFromCoords(coordinates);
   }
 
+  public static Geometry scale(Geometry geometry, double scaleX, double scaleY) {
+    return scaleGeom(geometry, Constructors.point(scaleX, scaleY));
+  }
+
+  public static Geometry scaleGeom(Geometry geometry, Geometry factor) {
+    return scaleGeom(geometry, factor, null);
+  }
+
+  public static Geometry scaleGeom(Geometry geometry, Geometry factor, Geometry origin) {
+    if (geometry == null || factor == null || geometry.isEmpty() || factor.isEmpty()) {
+      return geometry;
+    }
+
+    if (!factor.getGeometryType().equalsIgnoreCase(Geometry.TYPENAME_POINT)) {
+      throw new IllegalArgumentException("Scale factor geometry should be a Point type.");
+    }
+
+    Geometry resultGeom = null;
+    AffineTransformation scaleInstance = null;
+    Coordinate factorCoordinate = factor.getCoordinate();
+
+    if (origin == null || origin.isEmpty()) {
+      scaleInstance =
+          AffineTransformation.scaleInstance(factorCoordinate.getX(), factorCoordinate.getY());
+      resultGeom = scaleInstance.transform(geometry);
+    } else {
+      Coordinate falseOrigin = origin.getCoordinate();
+      scaleInstance =
+          AffineTransformation.scaleInstance(
+              factorCoordinate.getX(),
+              factorCoordinate.getY(),
+              falseOrigin.getX(),
+              falseOrigin.getY());
+      resultGeom = scaleInstance.transform(geometry);
+    }
+
+    return resultGeom;
+  }
+
   public static Geometry rotateX(Geometry geometry, double angle) {
     if (GeomUtils.isAnyGeomEmpty(geometry)) {
       return geometry;

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

@@ -3994,6 +3994,44 @@ public void points() throws ParseException {
     assertEquals("MULTIPOINT Z((0 0 1), (1 1 2), (2 2 3), (0 0 1))", result1);
   }
 
+  @Test
+  public void scale() throws ParseException {
+    Geometry geom = Constructors.geomFromWKT("POLYGON ((0 0, 0 1, 1 1, 1 0, 0 0))", 0);
+    Geometry actual = Functions.scale(geom, 3, 2);
+    String expected = "POLYGON ((0 0, 0 2, 3 2, 3 0, 0 0))";
+    assertEquals(expected, actual.toString());
+
+    geom = Constructors.geomFromWKT("LINESTRING(0 1, 1 0)", 0);
+    actual = Functions.scale(geom, 10, 5);
+    expected = "LINESTRING (0 5, 10 0)";
+    assertEquals(expected, actual.toString());
+
+    geom = Constructors.geomFromWKT("POLYGON ((0 0, 0 1.5, 1.5 1.5, 1.5 0, 0 0))", 1111);
+    actual = Functions.scaleGeom(geom, Constructors.point(1.8, 2.1));
+    expected = "POLYGON ((0 0, 0 3.1500000000000004, 2.7 3.1500000000000004, 2.7 0, 0 0))";
+    assertEquals(expected, actual.toString());
+    assertEquals(1111, actual.getSRID());
+
+    actual =
+        Functions.scaleGeom(geom, Constructors.point(3, 2), Constructors.point(0.32959, 0.796483));
+    expected =
+        "POLYGON ((-0.6591799999999999 -0.796483, -0.6591799999999999 2.2035169999999997, 3.84082 2.2035169999999997, 3.84082 -0.796483, -0.6591799999999999 -0.796483))";
+    assertEquals(expected, actual.toString());
+
+    // test to check Z and M ordinate preservation
+    geom = Constructors.geomFromWKT("POLYGON ((0 0 1, 0 1.5 2, 1.5 1.5 2, 1.5 0 3, 0 0 1))", 0);
+    String actualWKT = Functions.asWKT(Functions.scale(geom, 3, 2));
+    expected = "POLYGON Z((0 0 1, 0 3 2, 4.5 3 2, 4.5 0 3, 0 0 1))";
+    assertEquals(expected, actualWKT);
+
+    geom =
+        Constructors.geomFromWKT(
+            "POLYGON ZM((0 0 1 2, 0 1.5 2 2, 1.5 1.5 2 2, 1.5 0 3 2, 0 0 1 2))", 0);
+    actualWKT = Functions.asWKT(Functions.scale(geom, 3, 2));
+    expected = "POLYGON ZM((0 0 1 2, 0 3 2 2, 4.5 3 2 2, 4.5 0 3 2, 0 0 1 2))";
+    assertEquals(expected, actualWKT);
+  }
+
   @Test
   public void rotateX() throws ParseException {
     Geometry lineString = Constructors.geomFromEWKT("LINESTRING (50 160, 50 50, 100 50)");

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

@@ -100,6 +100,8 @@ public static UserDefinedFunction[] getFuncs() {
       new Functions.ST_FlipCoordinates(),
       new Functions.ST_GeoHash(),
       new Functions.ST_PointOnSurface(),
+      new Functions.ST_Scale(),
+      new Functions.ST_ScaleGeom(),
       new Functions.ST_ReducePrecision(),
       new Functions.ST_Reverse(),
       new Functions.ST_Rotate(),

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

@@ -1977,6 +1977,44 @@ public String eval(
     }
   }
 
+  public static class ST_Scale 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 scaleX,
+        @DataTypeHint(value = "Double") Double scaleY) {
+      Geometry geometry = (Geometry) o;
+      return org.apache.sedona.common.Functions.scale(geometry, scaleX, scaleY);
+    }
+  }
+
+  public static class ST_ScaleGeom 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(value = "RAW", bridgedTo = org.locationtech.jts.geom.Geometry.class)
+            Object o3) {
+      Geometry geometry = (Geometry) o1;
+      Geometry factor = (Geometry) o2;
+      Geometry origin = (Geometry) o3;
+      return org.apache.sedona.common.Functions.scaleGeom(geometry, factor, origin);
+    }
+
+    @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) {
+      Geometry geometry = (Geometry) o1;
+      Geometry factor = (Geometry) o2;
+      return org.apache.sedona.common.Functions.scaleGeom(geometry, factor);
+    }
+  }
+
   public static class ST_RotateX 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

@@ -2654,6 +2654,52 @@ public void testIsValidReason() {
     // standards
   }
 
+  @Test
+  public void testScale() {
+    Table tbl =
+        tableEnv.sqlQuery(
+            "SELECT ST_GeomFromWKT('POLYGON ((0 0, 2 0, 1 1, 2 2, 0 2, 1 1, 0 0))', 1010) AS geom");
+    Geometry actual =
+        (Geometry)
+            first(tbl.select(call(Functions.ST_Scale.class.getSimpleName(), $("geom"), 2, 3)))
+                .getField(0);
+    String expected = "POLYGON ((0 0, 4 0, 2 3, 4 6, 0 6, 2 3, 0 0))";
+    assertEquals(expected, actual.toString());
+    assertEquals(1010, actual.getSRID());
+  }
+
+  @Test
+  public void testScaleGeom() {
+    Table tbl =
+        tableEnv.sqlQuery(
+            "SELECT ST_GeomFromWKT('POLYGON ((0 0, 2 0, 1 1, 2 2, 0 2, 1 1, 0 0))', 1010) AS geom, ST_GeomFromWKT('POINT (2 3)') AS factor");
+    Geometry actual =
+        (Geometry)
+            first(
+                    tbl.select(
+                        call(Functions.ST_ScaleGeom.class.getSimpleName(), $("geom"), $("factor"))))
+                .getField(0);
+    String expected = "POLYGON ((0 0, 4 0, 2 3, 4 6, 0 6, 2 3, 0 0))";
+    assertEquals(expected, actual.toString());
+    assertEquals(1010, actual.getSRID());
+
+    tbl =
+        tableEnv.sqlQuery(
+            "SELECT ST_GeomFromWKT('POLYGON ((0 0, 2 0, 1 1, 2 2, 0 2, 1 1, 0 0))', 1010) AS geom, ST_GeomFromWKT('POINT (2 3)') AS factor, ST_GeomFromWKT('POINT (-1 0)') AS origin");
+    actual =
+        (Geometry)
+            first(
+                    tbl.select(
+                        call(
+                            Functions.ST_ScaleGeom.class.getSimpleName(),
+                            $("geom"),
+                            $("factor"),
+                            $("origin"))))
+                .getField(0);
+    expected = "POLYGON ((1 0, 5 0, 3 3, 5 6, 1 6, 3 3, 1 0))";
+    assertEquals(expected, actual.toString());
+  }
+
   @Test
   public void testRotateX() {
     Table tbl =

python/sedona/sql/st_functions.py

@@ -2298,6 +2298,40 @@ def ST_IsCollection(geometry: ColumnOrName) -> Column:
     return _call_st_function("ST_IsCollection", geometry)
 
 
+@validate_argument_types
+def ST_Scale(
+    geometry: ColumnOrName,
+    scaleX: Union[ColumnOrNameOrNumber, float],
+    scaleY: Union[ColumnOrNameOrNumber, float],
+) -> Column:
+    """Scale geometry with X and Y axis.
+
+    @param geometry:
+    @param scaleX:
+    @param scaleY:
+    @return:
+    """
+    return _call_st_function("ST_Scale", (geometry, scaleX, scaleY))
+
+
+@validate_argument_types
+def ST_ScaleGeom(
+    geometry: ColumnOrName, factor: ColumnOrName, origin: Optional[ColumnOrName] = None
+) -> Column:
+    """Scale geometry with the corodinates of factor geometry
+
+    @param geometry:
+    @param factor:
+    @param origin:
+    @return:
+    """
+    if origin is not None:
+        args = (geometry, factor, origin)
+    else:
+        args = (geometry, factor)
+    return _call_st_function("ST_ScaleGeom", args)
+
+
 @validate_argument_types
 def ST_RotateX(geometry: ColumnOrName, angle: Union[ColumnOrName, float]) -> Column:
     """Returns geometry rotated by the given angle in X axis

python/tests/sql/test_dataframe_api.py

@@ -845,6 +845,20 @@
         "",
         "LINESTRING (5 0, 4 0, 3 0, 2 0, 1 0, 0 0)",
     ),
+    (
+        stf.ST_Scale,
+        ("poly", 3, 2),
+        "poly_and_point",
+        "",
+        "POLYGON ((0 0, 0 2, 3 2, 3 0, 0 0))",
+    ),
+    (
+        stf.ST_ScaleGeom,
+        ("poly", "point"),
+        "poly_and_point",
+        "",
+        "POLYGON ((0 0, 0 2, 3 2, 3 0, 0 0))",
+    ),
     (
         stf.ST_RotateX,
         ("line", 10.0),
@@ -1316,6 +1330,8 @@
     (stf.ST_RemovePoint, ("", 1.0)),
     (stf.ST_RemoveRepeatedPoints, (None, None)),
     (stf.ST_Reverse, (None,)),
+    (stf.ST_Scale, (None, None, None)),
+    (stf.ST_ScaleGeom, (None, None, None)),
     (
         stf.ST_Rotate,
         (
@@ -1592,6 +1608,10 @@ def base_df(self, request):
             return TestDataFrameAPI.spark.sql(
                 "SELECT array(ST_GeomFromWKT('POLYGON ((-3 -3, 3 -3, 3 3, -3 3, -3 -3))'), ST_GeomFromWKT('POLYGON ((-2 1, 2 1, 2 4, -2 4, -2 1))')) as polys"
             )
+        elif request.param == "poly_and_point":
+            return TestDataFrameAPI.spark.sql(
+                "SELECT ST_GeomFromWKT('POLYGON ((0 0, 0 1, 1 1, 1 0, 0 0))') AS poly, ST_GeomFromWKT('POINT (3 2)') AS point"
+            )
         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"

python/tests/sql/test_function.py

@@ -1345,6 +1345,32 @@ def test_st_add_point(self):
         ]
         assert collected_geometries[0] == "LINESTRING (0 0, 1 1, 1 0, 21 52)"
 
+    def test_st_scale(self):
+        baseDf = self.spark.sql(
+            "SELECT ST_GeomFromWKT('LINESTRING (50 160, 50 50, 100 50)') AS geom"
+        )
+        actual = baseDf.selectExpr("ST_AsText(ST_Scale(geom, -10, 5))").first()[0]
+        expected = "LINESTRING (-500 800, -500 250, -1000 250)"
+        assert expected == actual
+
+    def test_st_scalegeom(self):
+        baseDf = self.spark.sql(
+            "SELECT ST_GeomFromWKT('POLYGON ((0 0, 0 1.5, 1.5 1.5, 1.5 0, 0 0))') AS geometry, ST_GeomFromWKT('POINT (1.8 2.1)') AS factor, ST_GeomFromWKT('POINT (0.32959 0.796483)') AS origin"
+        )
+        actual = baseDf.selectExpr("ST_AsText(ST_ScaleGeom(geometry, factor))").first()[
+            0
+        ]
+        expected = (
+            "POLYGON ((0 0, 0 3.1500000000000004, 2.7 3.1500000000000004, 2.7 0, 0 0))"
+        )
+        assert expected == actual
+
+        actual = baseDf.selectExpr(
+            "ST_AsText(ST_ScaleGeom(geometry, factor, origin))"
+        ).first()[0]
+        expected = "POLYGON ((-0.263672 -0.8761313000000002, -0.263672 2.2738687000000004, 2.436328 2.2738687000000004, 2.436328 -0.8761313000000002, -0.263672 -0.8761313000000002))"
+        assert expected == actual
+
     def test_st_rotate_x(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"

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

@@ -1237,6 +1237,27 @@ public void test_ST_Translate() {
         "GEOMETRYCOLLECTION Z(MULTIPOLYGON Z(((3 2 3, 3 3 3, 4 3 3, 4 2 3, 3 2 3)), ((3 4 3, 5 6 3, 5 7 3, 3 4 3))), POINT Z(3 3 4), LINESTRING ZEMPTY)");
   }
 
+  @Test
+  public void test_ST_Scale() {
+    registerUDF("ST_Scale", byte[].class, double.class, double.class);
+    verifySqlSingleRes(
+        "SELECT sedona.ST_AsText(sedona.ST_Scale(sedona.ST_GeomFromWKT('POLYGON ((0 0, 0 1, 1 1, 1 0, 0 0))'), 3, 2))",
+        "POLYGON ((0 0, 0 2, 3 2, 3 0, 0 0))");
+  }
+
+  @Test
+  public void test_ST_ScaleGeom() {
+    registerUDF("ST_ScaleGeom", byte[].class, byte[].class, byte[].class);
+    verifySqlSingleRes(
+        "SELECT sedona.ST_AsText(sedona.ST_ScaleGeom(sedona.ST_GeomFromWKT('POLYGON ((0 0, 0 1, 1 1, 1 0, 0 0))'), sedona.ST_Point(3, 2), sedona.ST_Point(1, 2)))",
+        "POLYGON ((-2 -2, -2 0, 1 0, 1 -2, -2 -2))");
+
+    registerUDF("ST_ScaleGeom", byte[].class, byte[].class);
+    verifySqlSingleRes(
+        "SELECT sedona.ST_AsText(sedona.ST_ScaleGeom(sedona.ST_GeomFromWKT('POLYGON ((0 0, 0 1, 1 1, 1 0, 0 0))'), sedona.ST_Point(3, 2)))",
+        "POLYGON ((0 0, 0 2, 3 2, 3 0, 0 0))");
+  }
+
   @Test
   public void test_ST_RotateX() {
     registerUDF("ST_RotateX", byte[].class, double.class);

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

@@ -1192,6 +1192,27 @@ public void test_ST_Translate() {
         "POINT(2 5)");
   }
 
+  @Test
+  public void test_ST_Scale() {
+    registerUDFV2("ST_Scale", String.class, double.class, double.class);
+    verifySqlSingleRes(
+        "SELECT ST_AsText(sedona.ST_Scale(ST_GeometryFromWKT('POLYGON ((0 0, 0 1, 1 1, 1 0, 0 0))'), 3, 2))",
+        "POLYGON((0 0,0 2,3 2,3 0,0 0))");
+  }
+
+  @Test
+  public void test_ST_ScaleGeom() {
+    registerUDFV2("ST_ScaleGeom", String.class, String.class, String.class);
+    verifySqlSingleRes(
+        "SELECT ST_AsText(sedona.ST_ScaleGeom(ST_GeometryFromWKT('POLYGON ((0 0, 0 1, 1 1, 1 0, 0 0))'), ST_Point(3, 2), ST_Point(1, 2)))",
+        "POLYGON((-2 -2,-2 0,1 0,1 -2,-2 -2))");
+
+    registerUDFV2("ST_ScaleGeom", String.class, String.class);
+    verifySqlSingleRes(
+        "SELECT ST_AsText(sedona.ST_ScaleGeom(ST_GeometryFromWKT('POLYGON ((0 0, 0 1, 1 1, 1 0, 0 0))'), ST_Point(3, 2)))",
+        "POLYGON((0 0,0 2,3 2,3 0,0 0))");
+  }
+
   @Test
   public void test_ST_RotateX() {
     registerUDFV2("ST_RotateX", String.class, double.class);

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

@@ -1290,6 +1290,28 @@ public static byte[] ST_Translate(byte[] geom, double deltaX, double deltaY, dou
         Functions.translate(GeometrySerde.deserialize(geom), deltaX, deltaY, deltaZ));
   }
 
+  @UDFAnnotations.ParamMeta(argNames = {"geometry", "scaleX", "scaleY"})
+  public static byte[] ST_Scale(byte[] geometry, double scaleX, double scaleY) {
+    return GeometrySerde.serialize(
+        Functions.scale(GeometrySerde.deserialize(geometry), scaleX, scaleY));
+  }
+
+  @UDFAnnotations.ParamMeta(argNames = {"geometry", "factor", "origin"})
+  public static byte[] ST_ScaleGeom(byte[] geometry, byte[] factor, byte[] origin) {
+    return GeometrySerde.serialize(
+        Functions.scaleGeom(
+            GeometrySerde.deserialize(geometry),
+            GeometrySerde.deserialize(factor),
+            GeometrySerde.deserialize(origin)));
+  }
+
+  @UDFAnnotations.ParamMeta(argNames = {"geometry", "factor"})
+  public static byte[] ST_ScaleGeom(byte[] geometry, byte[] factor) {
+    return GeometrySerde.serialize(
+        Functions.scaleGeom(
+            GeometrySerde.deserialize(geometry), GeometrySerde.deserialize(factor)));
+  }
+
   @UDFAnnotations.ParamMeta(argNames = {"geometry", "angle"})
   public static byte[] ST_RotateX(byte[] geometry, double angle) {
     return GeometrySerde.serialize(Functions.rotateX(GeometrySerde.deserialize(geometry), angle));