[SEDONA-159] Add Nth accessor functions to the Flink API

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

@@ -157,6 +157,23 @@ public static Geometry reverse(Geometry geometry) {
         return geometry.reverse();
     }
 
+    public static Geometry geometryN(Geometry geometry, int n) {
+        if (n < geometry.getNumGeometries()) {
+            return geometry.getGeometryN(n);
+        }
+        return null;
+    }
+
+    public static Geometry interiorRingN(Geometry geometry, int n) {
+        if (geometry instanceof Polygon) {
+            Polygon polygon = (Polygon) geometry;
+            if (n < polygon.getNumInteriorRing()) {
+                return polygon.getInteriorRingN(n);
+            }
+        }
+        return null;
+    }
+
     public static Geometry pointN(Geometry geometry, int n) {
         if(!(geometry instanceof LineString)) {
             return null;

docs/api/flink/Function.md

@@ -281,6 +281,36 @@ Result:
 +-----------------------------+
 ```
 
+## ST_GeometryN
+
+Introduction: Return the 0-based Nth geometry if the geometry is a GEOMETRYCOLLECTION, (MULTI)POINT, (MULTI)LINESTRING, MULTICURVE or (MULTI)POLYGON. Otherwise, return null
+
+Format: `ST_GeometryN(geom: geometry, n: Int)`
+
+Since: `v1.3.0`
+
+Example:
+```SQL
+SELECT ST_GeometryN(ST_GeomFromText('MULTIPOINT((1 2), (3 4), (5 6), (8 9))'), 1)
+```
+
+Output: `POINT (3 4)`
+
+## ST_InteriorRingN
+
+Introduction: Returns the Nth interior linestring ring of the polygon geometry. Returns NULL if the geometry is not a polygon or the given N is out of range
+
+Format: `ST_InteriorRingN(geom: geometry, n: Int)`
+
+Since: `v1.3.0`
+
+Example:
+```SQL
+SELECT ST_InteriorRingN(ST_GeomFromText('POLYGON((0 0, 0 5, 5 5, 5 0, 0 0), (1 1, 2 1, 2 2, 1 2, 1 1), (1 3, 2 3, 2 4, 1 4, 1 3), (3 3, 4 3, 4 4, 3 4, 3 3))'), 0)
+```
+
+Output: `LINEARRING (1 1, 2 1, 2 2, 1 2, 1 1)`
+
 ## ST_IsClosed
 
 Introduction: RETURNS true if the LINESTRING start and end point are the same.

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

@@ -45,6 +45,8 @@ public static UserDefinedFunction[] getFuncs() {
                 new Functions.ST_GeoHash(),
                 new Functions.ST_PointOnSurface(),
                 new Functions.ST_Reverse(),
+                new Functions.ST_GeometryN(),
+                new Functions.ST_InteriorRingN(),
                 new Functions.ST_PointN(),
                 new Functions.ST_ExteriorRing(),
                 new Functions.ST_AsEWKT(),

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

@@ -133,6 +133,22 @@ public Geometry eval(@DataTypeHint(value = "RAW", bridgedTo = org.locationtech.j
         }
     }
 
+    public static class ST_GeometryN 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, int n) {
+            Geometry geom = (Geometry) o;
+            return org.apache.sedona.common.Functions.geometryN(geom, n);
+        }
+    }
+
+    public static class ST_InteriorRingN 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, int n) {
+            Geometry geom = (Geometry) o;
+            return org.apache.sedona.common.Functions.interiorRingN(geom, n);
+        }
+    }
+
     public static class ST_PointN 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, int n) {

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

@@ -144,6 +144,22 @@ public void testReverse() {
         assertEquals("POLYGON ((-0.5 -0.5, 0.5 -0.5, 0.5 0.5, -0.5 0.5, -0.5 -0.5))", result.toString());
     }
 
+    @Test
+    public void testGeometryN() {
+        Table collectionTable = tableEnv.sqlQuery("SELECT ST_GeomFromWKT('GEOMETRYCOLLECTION(POINT(10 10), POINT(30 30), LINESTRING(15 15, 20 20))') AS collection");
+        Table resultTable = collectionTable.select(call(Functions.ST_GeometryN.class.getSimpleName(), $("collection"), 1));
+        Point point = (Point) first(resultTable).getField(0);
+        assertEquals("POINT (30 30)", point.toString());
+    }
+
+    @Test
+    public void testInteriorRingN() {
+        Table polygonTable = tableEnv.sqlQuery("SELECT ST_GeomFromText('POLYGON((7 9,8 7,11 6,15 8,16 6,17 7,17 10,18 12,17 14,15 15,11 15,10 13,9 12,7 9),(9 9,10 10,11 11,11 10,10 8,9 9),(12 14,15 14,13 11,12 14))') AS polygon");
+        Table resultTable = polygonTable.select(call(Functions.ST_InteriorRingN.class.getSimpleName(), $("polygon"), 1));
+        LinearRing linearRing = (LinearRing) first(resultTable).getField(0);
+        assertEquals("LINEARRING (12 14, 15 14, 13 11, 12 14)", linearRing.toString());
+    }
+
     @Test
     public void testPointN_positiveN() {
         int n = 1;

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

@@ -20,7 +20,6 @@ package org.apache.spark.sql.sedona_sql.expressions
 
 import org.apache.sedona.common.Functions
 import org.apache.sedona.sql.utils.GeometrySerializer
-import org.apache.spark.internal.Logging
 import org.apache.spark.sql.catalyst.InternalRow
 import org.apache.spark.sql.catalyst.expressions.codegen.CodegenFallback
 import org.apache.spark.sql.catalyst.expressions._
@@ -793,58 +792,15 @@ case class ST_ExteriorRing(inputExpressions: Seq[Expression])
 
 
 case class ST_GeometryN(inputExpressions: Seq[Expression])
-  extends Expression with CodegenFallback with Logging {
-  assert(inputExpressions.length == 2)
-
-  override def nullable: Boolean = true
-
-  override def eval(input: InternalRow): Any = {
-    val geometry = inputExpressions(0).toGeometry(input)
-    val n = inputExpressions(1).toInt(input)
-    geometry match {
-      case geom: Geometry => getNthGeom(geom, n)
-      case _ => null
-    }
-  }
-
-  private def getNthGeom(geom: Geometry, index: Int): GenericArrayData = {
-    val nthGeom = Try(geom.getGeometryN(index))
-    nthGeom match {
-      case Failure(exception) =>
-        logWarning(s"Geometry ${geom.toString}, Out Of index")
-        null
-      case Success(geom) => geom.toGenericArrayData
-    }
-  }
-
-  override def dataType: DataType = GeometryUDT
-
-  override def children: Seq[Expression] = inputExpressions
+  extends InferredBinaryExpression(Functions.geometryN) {
 
   protected def withNewChildrenInternal(newChildren: IndexedSeq[Expression]) = {
     copy(inputExpressions = newChildren)
   }
 }
 
 case class ST_InteriorRingN(inputExpressions: Seq[Expression])
-  extends Expression with CodegenFallback {
-  assert(inputExpressions.length == 2)
-
-  override def nullable: Boolean = true
-
-  override def eval(input: InternalRow): Any = {
-    val geometry = inputExpressions(0).toGeometry(input)
-    val n = inputExpressions(1).toInt(input)
-    geometry match {
-      case geom: Polygon => geom.getInteriorRingN(n)
-        .toGenericArrayData
-      case _ => null
-    }
-  }
-
-  override def dataType: DataType = GeometryUDT
-
-  override def children: Seq[Expression] = inputExpressions
+  extends InferredBinaryExpression(Functions.interiorRingN) {
 
   protected def withNewChildrenInternal(newChildren: IndexedSeq[Expression]) = {
     copy(inputExpressions = newChildren)