[SEDONA-701] Add ST_LineSegments (#1765)

* feat: add ST_LineSegments

* use instanceOf instead of try-catch

* fix spotless

* change arraylist to array

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

@@ -1156,6 +1156,36 @@ public static Geometry lineMerge(Geometry geometry) {
     return geometry.getFactory().createGeometryCollection();
   }
 
+  public static Geometry[] lineSegments(Geometry geometry, boolean lenient) {
+    if (!(geometry instanceof LineString)) {
+      if (lenient) {
+        return new Geometry[] {};
+      } else {
+        throw new IllegalArgumentException(
+            "Geometry is not a LineString. This function expects input geometry to be a LineString.");
+      }
+    }
+
+    LineString line = (LineString) geometry;
+    Coordinate[] coords = line.getCoordinates();
+    if (coords.length == 2 || coords.length == 0) {
+      return new Geometry[] {line};
+    }
+
+    GeometryFactory geometryFactory = geometry.getFactory();
+    Geometry[] resultArray = new Geometry[coords.length - 1];
+    for (int i = 1; i < coords.length; i++) {
+      resultArray[i - 1] =
+          geometryFactory.createLineString(new Coordinate[] {coords[i - 1], coords[i]});
+    }
+
+    return resultArray;
+  }
+
+  public static Geometry[] lineSegments(Geometry geometry) {
+    return lineSegments(geometry, true);
+  }
+
   public static Geometry minimumBoundingCircle(Geometry geometry, int quadrantSegments) {
     MinimumBoundingCircle minimumBoundingCircle = new MinimumBoundingCircle(geometry);
     Coordinate centre = minimumBoundingCircle.getCentre();

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

@@ -2431,6 +2431,45 @@ public void makeLineWithWrongType() {
         "ST_MakeLine only supports Point, MultiPoint and LineString geometries", e.getMessage());
   }
 
+  @Test
+  public void lineSegments() throws ParseException {
+    Geometry geom = Constructors.geomFromWKT("LINESTRING (0 0, 1 1, 2 2, 3 3, 3 4)", 0);
+    Geometry[] actual = Functions.lineSegments(geom, false);
+    int actualSize = actual.length;
+    int expectedSize = 4;
+    assertEquals(expectedSize, actualSize);
+
+    geom = Constructors.geomFromWKT("LINESTRING (0 0, 1 1)", 0);
+    actual = Functions.lineSegments(geom);
+    actualSize = actual.length;
+    expectedSize = 1;
+    assertEquals(expectedSize, actualSize);
+
+    geom = Constructors.geomFromWKT("LINESTRING (0 0, 1 1, 2 2, 3 3, 3 4, 4 4)", 4326);
+    actual = Functions.lineSegments(geom);
+    actualSize = actual.length;
+    expectedSize = 5;
+    assertEquals(expectedSize, actualSize);
+
+    // Check SRID
+    Geometry resultCheck = actual[0];
+    assertEquals(4326, resultCheck.getSRID());
+
+    geom = GEOMETRY_FACTORY.createLineString();
+    actual = Functions.lineSegments(geom);
+    String actualString = Arrays.toString(actual);
+    String expectedString = "[LINESTRING EMPTY]";
+    assertEquals(expectedString, actualString);
+
+    geom =
+        Constructors.geomFromWKT(
+            "POLYGON ((65.10498 18.625425, 62.182617 16.36231, 64.863281 16.40447, 62.006836 14.157882, 65.522461 14.008696, 65.10498 18.625425))",
+            0);
+    actual = Functions.lineSegments(geom, true);
+    actualSize = actual.length;
+    assertEquals(0, actualSize);
+  }
+
   @Test
   public void minimumBoundingRadius() {
     Point point = GEOMETRY_FACTORY.createPoint(new Coordinate(0, 0));

docs/api/flink/Function.md

@@ -2591,6 +2591,47 @@ Output:
 LINESTRING (-29 -27, -30 -29.7, -45 -33, -46 -32)
 ```
 
+## ST_LineSegments
+
+Introduction: This function transforms a LineString containing multiple coordinates into an array of LineStrings, each with precisely two coordinates. The `lenient` argument, true by default, prevents an exception from being raised if the input geometry is not a LineString.
+
+Format:
+
+`ST_LineSegments(geom: Geometry, lenient: Boolean)`
+
+`ST_LineSegments(geom: Geometry)`
+
+Since: `v1.7.1`
+
+SQL Example:
+
+```sql
+SELECT ST_LineSegments(
+        ST_GeomFromWKT('LINESTRING(0 0, 10 10, 20 20, 30 30, 40 40, 50 50)'),
+       false
+    )
+```
+
+Output:
+
+```
+[LINESTRING (0 0, 10 10), LINESTRING (10 10, 20 20), LINESTRING (20 20, 30 30), LINESTRING (30 30, 40 40), LINESTRING (40 40, 50 50)]
+```
+
+SQL Example:
+
+```sql
+SELECT ST_LineSegments(
+        ST_GeomFromWKT('POLYGON((0 0, 10 0, 10 10, 0 10, 0 0))')
+    )
+```
+
+Output:
+
+```
+[]
+```
+
 ## ST_LineSubstring
 
 Introduction: Return a linestring being a substring of the input one starting and ending at the given fractions of total 2d length. Second and third arguments are Double values between 0 and 1. This only works with LINESTRINGs.

docs/api/sql/Function.md

@@ -2710,6 +2710,47 @@ Output:
 LINESTRING (-29 -27, -30 -29.7, -45 -33, -46 -32)
 ```
 
+## ST_LineSegments
+
+Introduction: This function transforms a LineString containing multiple coordinates into an array of LineStrings, each with precisely two coordinates. The `lenient` argument, true by default, prevents an exception from being raised if the input geometry is not a LineString.
+
+Format:
+
+`ST_LineSegments(geom: Geometry, lenient: Boolean)`
+
+`ST_LineSegments(geom: Geometry)`
+
+Since: `v1.7.1`
+
+SQL Example:
+
+```sql
+SELECT ST_LineSegments(
+        ST_GeomFromWKT('LINESTRING(0 0, 10 10, 20 20, 30 30, 40 40, 50 50)'),
+       false
+    )
+```
+
+Output:
+
+```
+[LINESTRING (0 0, 10 10), LINESTRING (10 10, 20 20), LINESTRING (20 20, 30 30), LINESTRING (30 30, 40 40), LINESTRING (40 40, 50 50)]
+```
+
+SQL Example:
+
+```sql
+SELECT ST_LineSegments(
+        ST_GeomFromWKT('POLYGON((0 0, 10 0, 10 10, 0 10, 0 0))')
+    )
+```
+
+Output:
+
+```
+[]
+```
+
 ## ST_LineSubstring
 
 Introduction: Return a linestring being a substring of the input one starting and ending at the given fractions of total 2d length. Second and third arguments are Double values between 0 and 1. This only works with LINESTRINGs.

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

@@ -157,6 +157,7 @@ public static UserDefinedFunction[] getFuncs() {
       new Functions.ST_SetPoint(),
       new Functions.ST_LineFromMultiPoint(),
       new Functions.ST_LineMerge(),
+      new Functions.ST_LineSegments(),
       new Functions.ST_LineSubstring(),
       new Functions.ST_HasZ(),
       new Functions.ST_HasM(),

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

@@ -1130,6 +1130,24 @@ public Geometry eval(
     }
   }
 
+  public static class ST_LineSegments 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) {
+      Geometry geometry = (Geometry) o;
+      return org.apache.sedona.common.Functions.lineSegments(geometry);
+    }
+
+    @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 = "Boolean") Boolean lenient) {
+      Geometry geometry = (Geometry) o;
+      return org.apache.sedona.common.Functions.lineSegments(geometry, lenient);
+    }
+  }
+
   public static class ST_LineMerge 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

@@ -1610,6 +1610,30 @@ public void testLineMerge() {
     assertEquals("LINESTRING (10 160, 60 120, 120 140, 180 120)", result.toString());
   }
 
+  @Test
+  public void testLineSegments() {
+    Table baseTable =
+        tableEnv.sqlQuery(
+            "SELECT ST_GeomFromWKT('LINESTRING(120 140, 60 120, 30 20)') AS line, ST_GeomFromWKT('POLYGON ((0 0, 0 1, 1 0, 0 0))') AS poly");
+    Geometry[] result =
+        (Geometry[])
+            first(
+                    baseTable.select(
+                        call(Functions.ST_LineSegments.class.getSimpleName(), $("line"))))
+                .getField(0);
+    int actualSize = result.length;
+    int expectedSize = 2;
+    assertEquals(expectedSize, actualSize);
+
+    result =
+        (Geometry[])
+            first(
+                    baseTable.select(
+                        call(Functions.ST_LineSegments.class.getSimpleName(), $("poly"), true)))
+                .getField(0);
+    assertEquals(0, result.length);
+  }
+
   @Test
   public void testLineSubString() {
     Table table = tableEnv.sqlQuery("SELECT ST_GeomFromWKT('LINESTRING (0 0, 2 0)') AS line");

python/sedona/sql/st_functions.py

@@ -1069,6 +1069,24 @@ def ST_LineMerge(multi_line_string: ColumnOrName) -> Column:
     return _call_st_function("ST_LineMerge", multi_line_string)
 
 
+@validate_argument_types
+def ST_LineSegments(
+    geom: ColumnOrName, lenient: Optional[Union[ColumnOrName, bool]] = None
+) -> Column:
+    """
+    Convert multi-coordinate LineString into an array of LineStrings that contain exactly 2 points.
+
+    @param geom: input LineString geometry column.
+    @param lenient: suppresses exception
+    @return: array of LineStrings
+    """
+    args = (geom, lenient)
+    if lenient is None:
+        args = (geom,)
+
+    return _call_st_function("ST_LineSegments", args)
+
+
 @validate_argument_types
 def ST_LineSubstring(
     line_string: ColumnOrName,

python/tests/sql/test_dataframe_api.py

@@ -683,6 +683,14 @@
         "",
         "LINESTRING (0 0, 1 0, 1 1, 0 0)",
     ),
+    (stf.ST_LineSegments, ("line",), "linestring_geom", "array_size(geom)", 5),
+    (
+        stf.ST_LineSegments,
+        ("geom", True),
+        "polygon_unsimplified",
+        "array_size(geom)",
+        0,
+    ),
     (
         stf.ST_LineSubstring,
         ("line", 0.5, 1.0),

python/tests/sql/test_function.py

@@ -2162,6 +2162,19 @@ def test_st_buildarea(self):
             )
             assert areal_geom.take(1)[0][0] == expected_geom
 
+    def test_st_line_segments(self):
+        baseDf = self.spark.sql(
+            "SELECT ST_GeomFromWKT('LINESTRING(120 140, 60 120, 30 20)') AS line, ST_GeomFromWKT('POLYGON ((0 0, 0 1, 1 0, 0 0))') AS poly"
+        )
+        resultSize = baseDf.selectExpr(
+            "array_size(ST_LineSegments(line, false))"
+        ).first()[0]
+        expected = 2
+        assert expected == resultSize
+
+        resultSize = baseDf.selectExpr("array_size(ST_LineSegments(poly))").first()[0]
+        assert 0 == resultSize
+
     def test_st_line_from_multi_point(self):
         test_cases = {
             "'POLYGON((-1 0 0, 1 0 0, 0 0 1, 0 1 0, -1 0 0))'": None,

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

@@ -170,6 +170,7 @@ object Catalog {
     function[ST_IsPolygonCCW](),
     function[ST_ForcePolygonCCW](),
     function[ST_FlipCoordinates](),
+    function[ST_LineSegments](),
     function[ST_LineSubstring](),
     function[ST_LineInterpolatePoint](),
     function[ST_LineLocatePoint](),

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

@@ -730,6 +730,15 @@ case class ST_MMax(inputExpressions: Seq[Expression])
   }
 }
 
+case class ST_LineSegments(inputExpressions: Seq[Expression])
+    extends InferredExpression(
+      inferrableFunction2(Functions.lineSegments),
+      inferrableFunction1(Functions.lineSegments)) {
+  protected def withNewChildrenInternal(newChildren: IndexedSeq[Expression]) = {
+    copy(inputExpressions = newChildren)
+  }
+}
+
 /**
  * Return a linestring being a substring of the input one starting and ending at the given
  * fractions of total 2d length. Second and third arguments are Double values between 0 and 1.

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

@@ -333,6 +333,13 @@ object st_functions extends DataFrameAPI {
   def ST_LineMerge(multiLineString: String): Column =
     wrapExpression[ST_LineMerge](multiLineString)
 
+  def ST_LineSegments(geom: Column): Column = wrapExpression[ST_LineSegments](geom)
+  def ST_LineSegments(geom: String): Column = wrapExpression[ST_LineSegments](geom)
+  def ST_LineSegments(geom: Column, lenient: Column): Column =
+    wrapExpression[ST_LineSegments](geom, lenient)
+  def ST_LineSegments(geom: String, lenient: Boolean): Column =
+    wrapExpression[ST_LineSegments](geom, lenient)
+
   def ST_LineSubstring(lineString: Column, startFraction: Column, endFraction: Column): Column =
     wrapExpression[ST_LineSubstring](lineString, startFraction, endFraction)
   def ST_LineSubstring(lineString: String, startFraction: Double, endFraction: Double): Column =

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

@@ -59,6 +59,7 @@ class PreserveSRIDSuite extends TestBaseScala with TableDrivenPropertyChecks {
       ("ST_StartPoint(geom3)", 1000),
       ("ST_Snap(geom3, geom3, 0.1)", 1000),
       ("ST_Boundary(geom1)", 1000),
+      ("ST_LineSegments(geom3)[0]", 1000),
       ("ST_LineSubstring(geom3, 0.1, 0.9)", 1000),
       ("ST_LineInterpolatePoint(geom3, 0.1)", 1000),
       ("ST_EndPoint(geom3)", 1000),

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

@@ -1450,6 +1450,22 @@ class dataFrameAPITestScala extends TestBaseScala {
       assert(actualRadius == expectedRadius)
     }
 
+    it("Passed ST_LineSegments") {
+      val baseDf = sparkSession.sql(
+        "SELECT ST_GeomFromWKT('LINESTRING(120 140, 60 120, 30 20)') AS line, ST_GeomFromWKT('POLYGON ((0 0, 0 1, 1 0, 0 0))') AS poly")
+      var resultSize = baseDf
+        .select(ST_LineSegments("line", false))
+        .first()
+        .getAs[WrappedArray[Geometry]](0)
+        .length
+      val expected = 2
+      assertEquals(expected, resultSize)
+
+      resultSize =
+        baseDf.select(ST_LineSegments("poly")).first().getAs[WrappedArray[Geometry]](0).length
+      assertEquals(0, resultSize)
+    }
+
     it("Passed ST_LineSubstring") {
       val baseDf = sparkSession.sql("SELECT ST_GeomFromWKT('LINESTRING (0 0, 2 0)') AS line")
       val df = baseDf.select(ST_LineSubstring("line", 0.5, 1.0))

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

@@ -2348,6 +2348,17 @@ class functionTestScala
       .toList should contain theSameElementsAs List(0, 1, 1)
   }
 
+  it("Should pass ST_LineSegments") {
+    val baseDf = sparkSession.sql(
+      "SELECT ST_GeomFromWKT('LINESTRING(120 140, 60 120, 30 20)') AS line, ST_GeomFromWKT('POLYGON ((0 0, 0 1, 1 0, 0 0))') AS poly")
+    var resultSize = baseDf.selectExpr("array_size(ST_LineSegments(line, false))").first().get(0)
+    val expected = 2
+    assertEquals(expected, resultSize)
+
+    resultSize = baseDf.selectExpr("array_size(ST_LineSegments(poly))").first().get(0)
+    assertEquals(0, resultSize)
+  }
+
   it("Should pass ST_LineSubstring") {
     Given("Sample geometry dataframe")
     val geometryTable = Seq("LINESTRING(25 50, 100 125, 150 190)")