[SEDONA-342] Implement ST_Polygon (#944)

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

@@ -733,6 +733,14 @@ public static Geometry makePolygon(Geometry shell, Geometry[] holes) {
         }
     }
 
+    public static Geometry makepolygonWithSRID(Geometry lineString, Integer srid) {
+        Geometry geom = makePolygon(lineString, null);
+        if(geom != null) {
+            geom.setSRID(srid);
+        }
+        return geom;
+    }
+
     public static Geometry createMultiGeometry(Geometry[] geometries) {
         if (geometries.length > 1){
             return GEOMETRY_FACTORY.buildGeometry(Arrays.asList(geometries));

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

@@ -529,6 +529,21 @@ public void geometricMedianFailConverge() {
         assertEquals("Median failed to converge within 1.0E-06 after 5 iterations.", e.getMessage());
     }
 
+    @Test
+    public void makepolygonWithSRID() {
+        Geometry lineString1 = GEOMETRY_FACTORY.createLineString(coordArray(0, 0, 1, 1, 1, 0, 0, 0));
+        Geometry actual1 = Functions.makepolygonWithSRID(lineString1, 4326);
+        Geometry expected1 = GEOMETRY_FACTORY.createPolygon(coordArray(0, 0, 1, 1, 1, 0, 0, 0));
+        assertEquals(expected1.toText(), actual1.toText());
+        assertEquals(4326, actual1.getSRID());
+
+        Geometry lineString2 = GEOMETRY_FACTORY.createLineString(coordArray3d(75, 29, 1, 77, 29, 2, 77, 29, 3, 75, 29, 1));
+        Geometry actual2 = Functions.makepolygonWithSRID(lineString2, 123);
+        Geometry expected2 = GEOMETRY_FACTORY.createPolygon(coordArray3d(75, 29, 1, 77, 29, 2, 77, 29, 3, 75, 29, 1));
+        assertEquals(expected2.toText(), actual2.toText());
+        assertEquals(123, actual2.getSRID());
+    }
+
     @Test
     public void haversineDistance() {
         // Basic check

docs/api/flink/Function.md

@@ -1883,6 +1883,26 @@ FROM df
 
     Output: `POINT Z(0 0 1)`
 
+## ST_Polygon
+
+Introduction: Function to create a polygon built from the given LineString and sets the spatial reference system from the srid
+
+Format: `ST_Polygon(geom: geometry, srid: integer)`
+
+Since: `v1.5.0`
+
+Example:
+
+```sql
+SELECT ST_AsText( ST_Polygon(ST_GeomFromEWKT('LINESTRING(75 29 1, 77 29 2, 77 29 3, 75 29 1)'), 4326) );
+```
+
+Output:
+
+```
+POLYGON((75 29 1, 77 29 2, 77 29 3, 75 29 1))
+```
+
 ## ST_ReducePrecision
 
 Introduction: Reduce the decimals places in the coordinates of the geometry to the given number of decimal places. The last decimal place will be rounded. 

docs/api/sql/Function.md

@@ -1886,6 +1886,26 @@ SELECT ST_AsText(ST_PointOnSurface(ST_GeomFromText('LINESTRING(0 5 1, 0 0 1, 0 1
 
 ```
 
+## ST_Polygon
+
+Introduction: Function to create a polygon built from the given LineString and sets the spatial reference system from the srid
+
+Format: `ST_Polygon(geom: geometry, srid: integer)`
+
+Since: `v1.5.0`
+
+Example:
+
+```sql
+SELECT ST_AsText( ST_Polygon(ST_GeomFromEWKT('LINESTRING(75 29 1, 77 29 2, 77 29 3, 75 29 1)'), 4326) );
+```
+
+Output:
+
+```
+POLYGON((75 29 1, 77 29 2, 77 29 3, 75 29 1))
+```
+
 ## ST_ReducePrecision
 
 Introduction: Reduce the decimals places in the coordinates of the geometry to the given number of decimal places. The last decimal place will be rounded. This function was called ST_PrecisionReduce in versions prior to v1.5.0.

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

@@ -115,6 +115,7 @@ public static UserDefinedFunction[] getFuncs() {
                 new Functions.ST_LineMerge(),
                 new Functions.ST_LineSubstring(),
                 new Functions.ST_MakeLine(),
+                new Functions.ST_Polygon(),
                 new Functions.ST_MakePolygon(),
                 new Functions.ST_MakeValid(),
                 new Functions.ST_MinimumBoundingCircle(),

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

@@ -719,6 +719,15 @@ public Geometry eval(@DataTypeHint(value = "RAW", bridgedTo = org.locationtech.j
         }
     }
 
+    public static class ST_Polygon 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("Integer") Integer srid) {
+            Geometry linestring = (Geometry) o1;
+            return org.apache.sedona.common.Functions.makepolygonWithSRID(linestring, srid);
+        }
+    }
+
     public static class ST_MakeValid 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

@@ -794,6 +794,15 @@ public void testMakeLine() {
         assertEquals("LINESTRING (0 0, 1 1)", result.toString());
     }
 
+    @Test 
+    public void testPolygon() {
+        Table table = tableEnv.sqlQuery("SELECT ST_GeomFromWKT('LINESTRING (0 0, 1 0, 1 1, 0 0)') AS line");
+        table = table.select(call(Functions.ST_Polygon.class.getSimpleName(), $("line"), 4236));
+        Geometry result = (Geometry) first(table).getField(0);
+        assertEquals("POLYGON ((0 0, 1 0, 1 1, 0 0))", result.toString());
+        assertEquals(4236, result.getSRID());
+    }
+
     @Test 
     public void testMakePolygon() {
         Table table = tableEnv.sqlQuery("SELECT ST_GeomFromWKT('LINESTRING (0 0, 1 0, 1 1, 0 0)') AS line");

python/sedona/sql/st_functions.py

@@ -76,6 +76,7 @@
     "ST_LineMerge",
     "ST_LineSubstring",
     "ST_MakeLine",
+    "ST_Polygon"
     "ST_MakePolygon",
     "ST_MakeValid",
     "ST_MinimumBoundingCircle",
@@ -827,6 +828,19 @@ def ST_MakeLine(geom1: ColumnOrName, geom2: Optional[ColumnOrName] = None) -> Co
     args = (geom1,) if geom2 is None else (geom1, geom2)
     return _call_st_function("ST_MakeLine", args)
 
+@validate_argument_types
+def ST_Polygon(line_string: ColumnOrName, srid: ColumnOrNameOrNumber) -> Column:
+    """Create a polygon built from the given LineString and sets the spatial reference system from the srid.
+
+    :param line_string: Closed linestring geometry column that describes the exterior ring of the polygon.
+    :type line_string: ColumnOrName
+    :param srid: Spatial reference system identifier.
+    :type srid: ColumnOrNameOrNumber
+    :return: Polygon geometry column created from the input linestring.
+    :rtype: Column
+    """
+    return _call_st_function("ST_Polygon", (line_string, srid))
+
 @validate_argument_types
 def ST_MakePolygon(line_string: ColumnOrName, holes: Optional[ColumnOrName] = None) -> Column:
     """Create a polygon geometry from a linestring describing the exterior ring as well as an array of linestrings describing holes.

python/tests/sql/test_dataframe_api.py

@@ -119,6 +119,7 @@
     (stf.ST_LineSubstring, ("line", 0.5, 1.0), "linestring_geom", "", "LINESTRING (2.5 0, 3 0, 4 0, 5 0)"),
     (stf.ST_MakeValid, ("geom",), "invalid_geom", "", "MULTIPOLYGON (((1 5, 3 3, 1 1, 1 5)), ((5 3, 7 5, 7 1, 5 3)))"),
     (stf.ST_MakeLine, ("line1", "line2"), "two_lines", "", "LINESTRING (0 0, 1 1, 0 0, 3 2)"),
+    (stf.ST_Polygon, ("geom", 4236), "closed_linestring_geom", "", "POLYGON ((0 0, 1 0, 1 1, 0 0))"),
     (stf.ST_MakePolygon, ("geom",), "closed_linestring_geom", "", "POLYGON ((0 0, 1 0, 1 1, 0 0))"),
     (stf.ST_MinimumBoundingCircle, ("line", 8), "linestring_geom", "ST_ReducePrecision(geom, 2)", "POLYGON ((4.95 -0.49, 4.81 -0.96, 4.58 -1.39, 4.27 -1.77, 3.89 -2.08, 3.46 -2.31, 2.99 -2.45, 2.5 -2.5, 2.01 -2.45, 1.54 -2.31, 1.11 -2.08, 0.73 -1.77, 0.42 -1.39, 0.19 -0.96, 0.05 -0.49, 0 0, 0.05 0.49, 0.19 0.96, 0.42 1.39, 0.73 1.77, 1.11 2.08, 1.54 2.31, 2.01 2.45, 2.5 2.5, 2.99 2.45, 3.46 2.31, 3.89 2.08, 4.27 1.77, 4.58 1.39, 4.81 0.96, 4.95 0.49, 5 0, 4.95 -0.49))"),
     (stf.ST_MinimumBoundingCircle, ("line", 2), "linestring_geom", "ST_ReducePrecision(geom, 2)", "POLYGON ((4.27 -1.77, 2.5 -2.5, 0.73 -1.77, 0 0, 0.73 1.77, 2.5 2.5, 4.27 1.77, 5 0, 4.27 -1.77))"),

python/tests/sql/test_function.py

@@ -829,6 +829,30 @@ def test_st_make_line(self):
         for actual, expected in result:
             assert actual == expected
 
+    def test_st_polygon(self):
+        # Given
+        geometry_df = self.spark.createDataFrame(
+            [
+                ["POINT(21 52)", 4238, None],
+                ["POLYGON ((35 10, 45 45, 15 40, 10 20, 35 10), (20 30, 35 35, 30 20, 20 30))", 4237, None],
+                ["LINESTRING (0 0, 0 1, 1 0, 0 0)", 4236, "POLYGON ((0 0, 0 1, 1 0, 0 0))"]
+            ]
+        ).selectExpr("ST_GeomFromText(_1) AS geom", "_2 AS srid", "_3 AS expected")
+
+        # When calling st_Polygon
+        geom_poly = geometry_df.withColumn("polygon", expr("ST_Polygon(geom, srid)"))
+
+        # Then only based on closed linestring geom is created
+        geom_poly.filter("polygon IS NOT NULL").selectExpr("ST_AsText(polygon)", "expected"). \
+            show()
+        result = geom_poly.filter("polygon IS NOT NULL").selectExpr("ST_AsText(polygon)", "expected"). \
+            collect()
+
+        assert result.__len__() == 1
+
+        for actual, expected in result:
+            assert actual == expected
+
     def test_st_make_polygon(self):
         # Given
         geometry_df = self.spark.createDataFrame(

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

@@ -124,6 +124,7 @@ object Catalog {
     function[ST_SubDivideExplode](),
     function[ST_SubDivide](),
     function[ST_MakeLine](),
+    function[ST_Polygon](),
     function[ST_MakePolygon](null),
     function[ST_GeoHash](),
     function[ST_GeomFromGeoHash](null),

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

@@ -696,6 +696,14 @@ case class ST_MakeLine(inputExpressions: Seq[Expression])
   }
 }
 
+case class ST_Polygon(inputExpressions: Seq[Expression])
+  extends InferredExpression(Functions.makepolygonWithSRID _) {
+
+  protected def withNewChildrenInternal(newChildren: IndexedSeq[Expression]) = {
+    copy(inputExpressions = newChildren)
+  }
+}
+
 case class ST_MakePolygon(inputExpressions: Seq[Expression])
   extends InferredExpression(InferrableFunction.allowRightNull(Functions.makePolygon)) {
 

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

@@ -173,6 +173,9 @@ object st_functions extends DataFrameAPI {
   def ST_MakeLine(geom1: Column, geom2: Column): Column = wrapExpression[ST_MakeLine](geom1, geom2)
   def ST_MakeLine(geom1: String, geom2: String): Column = wrapExpression[ST_MakeLine](geom1, geom2)
 
+  def ST_Polygon(lineString: Column, srid: Column): Column = wrapExpression[ST_Polygon](lineString, srid)
+  def ST_Polygon(lineString: String, srid: Integer): Column = wrapExpression[ST_Polygon](lineString, srid)
+
   def ST_MakePolygon(lineString: Column): Column = wrapExpression[ST_MakePolygon](lineString, null)
   def ST_MakePolygon(lineString: String): Column = wrapExpression[ST_MakePolygon](lineString, null)
   def ST_MakePolygon(lineString: Column, holes: Column): Column = wrapExpression[ST_MakePolygon](lineString, holes)

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

@@ -317,6 +317,15 @@ class dataFrameAPITestScala extends TestBaseScala {
       assert(actualResult == expectedResult)
     }
 
+    it("Passed ST_Polygon") {
+      val invalidDf = sparkSession.sql("SELECT ST_GeomFromWKT('LINESTRING (0 0, 1 0, 1 1, 0 0)') AS geom")
+      val df = invalidDf.select(ST_Polygon("geom", 4236))
+      val actualResult = df.take(1)(0).get(0).asInstanceOf[Geometry]
+      val expectedResult = "POLYGON ((0 0, 1 0, 1 1, 0 0))"
+      assert(actualResult.toText()  == expectedResult)
+      assert(actualResult.getSRID() == 4236)
+    }
+
     it("Passed `ST_MakePolygon`") {
       val invalidDf = sparkSession.sql("SELECT ST_GeomFromWKT('LINESTRING (0 0, 1 0, 1 1, 0 0)') AS geom")
       val df = invalidDf.select(ST_MakePolygon("geom"))

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

@@ -337,6 +337,15 @@ class functionTestScala extends TestBaseScala with Matchers with GeometrySample
       assert(testtable.take(1)(0).get(0).asInstanceOf[Geometry].toText.equals("LINESTRING (1 2, 3 4)"))
     }
 
+    it("Passed ST_Polygon") {
+
+      var testtable = sparkSession.sql(
+        "SELECT ST_Polygon(ST_GeomFromText('LINESTRING(75.15 29.53,77 29,77.6 29.5, 75.15 29.53)'), 4326)"
+      )
+      assert(testtable.take(1)(0).get(0).asInstanceOf[Geometry].toText  == "POLYGON ((75.15 29.53, 77 29, 77.6 29.5, 75.15 29.53))")
+      assert(testtable.take(1)(0).get(0).asInstanceOf[Geometry].getSRID == 4326)
+    }
+
     it("Passed ST_MakeValid On Invalid Polygon") {
 
       val df = sparkSession.sql("SELECT ST_GeomFromWKT('POLYGON((1 5, 1 1, 3 3, 5 3, 7 1, 7 5, 5 3, 3 3, 1 5))') AS polygon")