[SEDONA-293] Implement ST_IsCollection

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

@@ -988,6 +988,14 @@ public static double frechetDistance(Geometry g1, Geometry g2) {
         return GeomUtils.getFrechetDistance(g1, g2);
     }
 
+    public static boolean isCollection(Geometry geometry) {
+        String geoType = geometry.getGeometryType();
+        return Geometry.TYPENAME_GEOMETRYCOLLECTION.equalsIgnoreCase(geoType) ||
+                Geometry.TYPENAME_MULTIPOINT.equalsIgnoreCase(geoType) ||
+                Geometry.TYPENAME_MULTIPOLYGON.equalsIgnoreCase(geoType) ||
+                Geometry.TYPENAME_MULTILINESTRING.equalsIgnoreCase(geoType);
+    }
+
     public static Geometry boundingDiagonal(Geometry geometry) {
         if (geometry.isEmpty()) {
             return GEOMETRY_FACTORY.createLineString();

docs/api/flink/Function.md

@@ -1087,6 +1087,20 @@ Example:
 SELECT ST_IsClosed(ST_GeomFromText('LINESTRING(0 0, 1 1, 1 0)'))
 ```
 
+## ST_IsCollection
+
+Introduction: Returns `TRUE` if the argument type is a geometry collection type.
+
+Format: `ST_IsCollection(geom: geometry)`
+
+Since: `v1.5.0`
+
+Example:
+
+```sql
+SELECT ST_IsCollection(ST_GeomFromText('MULTIPOINT(0 0), (6 6)'))
+```
+
 ## ST_IsEmpty
 
 Introduction: Test if a geometry is empty geometry

docs/api/sql/Function.md

@@ -1082,6 +1082,20 @@ SELECT ST_IsClosed(ST_GeomFromText('LINESTRING(0 0, 1 1, 1 0)'))
 
 Output: `false`
 
+## ST_IsCollection
+
+Introduction: Returns `TRUE` if the argument type is a geometry collection type.
+
+Format: `ST_IsCollection(geom: geometry)`
+
+Since: `v1.5.0`
+
+Example:
+
+```sql
+SELECT ST_IsCollection(ST_GeomFromText('MULTIPOINT(0 0), (6 6)'))
+```
+
 ## ST_IsEmpty
 
 Introduction: Test if a geometry is empty geometry

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

@@ -134,6 +134,7 @@ public static UserDefinedFunction[] getFuncs() {
                 new Functions.ST_Angle(),
                 new Functions.ST_Degrees(),
                 new Functions.ST_HausdorffDistance(),
+                new Functions.ST_IsCollection(),
                 new Functions.ST_CoordDim(),
         };
     }

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

@@ -13,6 +13,7 @@
  */
 package org.apache.sedona.flink.expressions;
 
+import org.apache.calcite.interpreter.Scalar;
 import org.apache.commons.lang3.tuple.Pair;
 import org.apache.flink.table.annotation.DataTypeHint;
 import org.apache.flink.table.annotation.InputGroup;
@@ -948,6 +949,14 @@ public Integer eval(@DataTypeHint(value = "RAW", bridgedTo = org.locationtech.jt
         }
     }
 
+    public static class ST_IsCollection extends ScalarFunction {
+        @DataTypeHint("Boolean")
+        public boolean eval(@DataTypeHint(value = "RAW", bridgedTo = org.locationtech.jts.geom.Geometry.class) Object o) {
+            Geometry geom = (Geometry) o;
+            return org.apache.sedona.common.Functions.isCollection(geom);
+        }
+    }
+
     public static class ST_Angle extends ScalarFunction {
 
         @DataTypeHint("Double")

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

@@ -1047,6 +1047,22 @@ public void testHausdorffDistance() {
         assertEquals(expectedDefault, actualDefault);
     }
 
+    @Test
+    public void testIsCollectionForCollection() {
+        Table collectionTable = tableEnv.sqlQuery("SELECT ST_GeomFromWKT('GEOMETRYCOLLECTION(POINT(2 3), POINT(4 6), LINESTRING(15 15, 20 20))') AS collection");
+        Table resultTable = collectionTable.select(call(Functions.ST_IsCollection.class.getSimpleName(), $("collection")));
+        boolean result = (boolean) first(resultTable).getField(0);
+        assertTrue(result);
+    }
+
+    @Test
+    public void testIsCollectionForNotCollection() {
+        Table collectionTable = tableEnv.sqlQuery("SELECT ST_GeomFromWKT('POINT(10 10)') AS collection");
+        Table resultTable = collectionTable.select(call(Functions.ST_IsCollection.class.getSimpleName(), $("collection")));
+        boolean result = (boolean) first(resultTable).getField(0);
+        assertFalse(result);
+    }
+
     @Test
     public void testCoordDimFor2D() {
         Table polygonTable = tableEnv.sqlQuery("SELECT ST_GeomFromWKT('POINT(3 7)') AS " + polygonColNames[0]);

python/sedona/sql/st_functions.py

@@ -119,6 +119,7 @@
     "ST_Degrees",
     "ST_FrechetDistance",
     "ST_CoordDim",
+    "ST_IsCollection",
     "ST_Affine",
     "ST_BoundingDiagonal"
 ]
@@ -1434,3 +1435,13 @@ def ST_CoordDim(geometry: ColumnOrName) -> Column:
     """
     return _call_st_function("ST_CoordDim", geometry)
 
+@validate_argument_types
+def ST_IsCollection(geometry: ColumnOrName) -> Column:
+    """Check if the geometry is of GeometryCollection type.
+
+    :param geometry: Column for geometry collection
+    :type geometry: ColumnOrName
+    :return: True if geometry is a collection of geometries.
+    :rtype: Column
+    """
+    return _call_st_function("ST_IsCollection", geometry)

python/tests/sql/test_dataframe_api.py

@@ -104,6 +104,7 @@
     (stf.ST_HausdorffDistance, ("point", "line",), "point_and_line", "", 5.0990195135927845),
     (stf.ST_InteriorRingN, ("geom", 0), "geom_with_hole", "", "LINESTRING (1 1, 2 2, 2 1, 1 1)"),
     (stf.ST_Intersection, ("a", "b"), "overlapping_polys", "", "POLYGON ((2 0, 1 0, 1 1, 2 1, 2 0))"),
+    (stf.ST_IsCollection, ("geom",), "geom_collection", "", True),
     (stf.ST_IsClosed, ("geom",), "closed_linestring_geom", "", True),
     (stf.ST_IsEmpty, ("geom",), "empty_geom", "", True),
     (stf.ST_IsRing, ("line",), "linestring_geom", "", False),

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

@@ -111,6 +111,7 @@ object Catalog {
     function[ST_Dump](),
     function[ST_DumpPoints](),
     function[ST_IsClosed](),
+    function[ST_IsCollection](),
     function[ST_NumInteriorRings](),
     function[ST_AddPoint](-1),
     function[ST_RemovePoint](-1),

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

@@ -1083,3 +1083,15 @@ case class ST_CoordDim(inputExpressions: Seq[Expression])
   }
 }
 
+/**
+ * Returns True if geometry is a collection of geometries
+ *
+ * @param inputExpressions
+ */
+case class ST_IsCollection(inputExpressions: Seq[Expression])
+  extends InferredExpression(Functions.isCollection _) {
+  protected def withNewChildrenInternal(newChildren: IndexedSeq[Expression]): Expression = {
+    copy(inputExpressions = newChildren)
+  }
+}
+

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

@@ -384,5 +384,9 @@ object st_functions extends DataFrameAPI {
 
   def ST_CoordDim(geometry: String): Column = wrapExpression[ST_CoordDim](geometry)
 
+  def ST_IsCollection(geometry: Column): Column = wrapExpression[ST_IsCollection](geometry)
+
+  def ST_IsCollection(geometry: String): Column = wrapExpression[ST_IsCollection](geometry)
+
 }
 

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

@@ -2176,6 +2176,12 @@ class functionTestScala extends TestBaseScala with Matchers with GeometrySample
     assert(test.take(1)(0).get(0).asInstanceOf[Int] == 4)
   }
 
+  it("Should pass ST_IsCollection") {
+    val test = sparkSession.sql("SELECT ST_IsCollection(ST_GeomFromText('GEOMETRYCOLLECTION(POINT(2 3), POINT(4 6), LINESTRING(15 15, 20 20))'))")
+    val expected = true
+    assert(test.take(1)(0).get(0).asInstanceOf[Boolean] == expected)
+  }
+
   it ("should pass GeometryType") {
     val geomTestCases = Map (
       ("'POINT (51.3168 -0.56)'") -> "'POINT'",