[SEDONA-114] Implement ST_MakeLine (#943)

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

@@ -684,6 +684,28 @@ public static Geometry[] subDivide(Geometry geometry, int maxVertices) {
         return GeometrySubDivider.subDivide(geometry, maxVertices);
     }
 
+    public static Geometry makeLine(Geometry geom1, Geometry geom2) {
+       Geometry[] geoms = new Geometry[]{geom1, geom2};
+       return makeLine(geoms);
+    }
+
+    public static Geometry makeLine(Geometry[] geoms) {
+        ArrayList<Coordinate> coordinates = new ArrayList<>();      
+        for (Geometry geom : geoms) {
+            if (geom instanceof Point || geom instanceof MultiPoint || geom instanceof LineString) {
+                for (Coordinate coord : geom.getCoordinates()) {
+                    coordinates.add(coord);
+                }
+            }
+            else {
+                throw new IllegalArgumentException("ST_MakeLine only supports Point, MultiPoint and LineString geometries");
+            }
+        }
+
+        Coordinate[] coords = coordinates.toArray(new Coordinate[0]);
+        return GEOMETRY_FACTORY.createLineString(coords);
+    }
+
     public static Geometry makePolygon(Geometry shell, Geometry[] holes) {
         try {
             if (holes != null) {

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

@@ -763,6 +763,57 @@ public void force3DHybridGeomCollectionDefaultValue() {
         assertEquals(wktWriter3D.write(expectedLineString3D), wktWriter3D.write(actualGeometryCollection.getGeometryN(0).getGeometryN(3)));
     }
 
+    @Test
+    public void makeLine() {
+        Point point1 = GEOMETRY_FACTORY.createPoint(new Coordinate(0, 0));
+        Point point2 = GEOMETRY_FACTORY.createPoint(new Coordinate(1, 1));
+        String actual = Functions.makeLine(point1, point2).toText();
+        assertEquals("LINESTRING (0 0, 1 1)", actual);
+
+        MultiPoint multiPoint1 = GEOMETRY_FACTORY.createMultiPointFromCoords(coordArray(0.5, 0.5, 1.0, 1.0));
+        MultiPoint multiPoint2 = GEOMETRY_FACTORY.createMultiPointFromCoords(coordArray(0.5, 0.5, 2, 2));
+        String actual2 = Functions.makeLine(multiPoint1, multiPoint2).toText();
+        assertEquals("LINESTRING (0.5 0.5, 1 1, 0.5 0.5, 2 2)", actual2);
+
+        LineString line1 = GEOMETRY_FACTORY.createLineString(coordArray(0, 0, 1, 1));
+        LineString line2 = GEOMETRY_FACTORY.createLineString(coordArray(2, 2, 3, 3));
+        Geometry[] geoms = new Geometry[]{line1, line2};
+        String actual3 = Functions.makeLine(geoms).toText();
+        assertEquals("LINESTRING (0 0, 1 1, 2 2, 3 3)", actual3);
+    }
+
+    @Test
+    public void makeLine3d() {
+        WKTWriter wktWriter3D = new WKTWriter(3);
+        Point point1 = GEOMETRY_FACTORY.createPoint(new Coordinate(1, 1, 1));
+        Point point2 = GEOMETRY_FACTORY.createPoint(new Coordinate(2, 2, 2));
+        LineString acutalLinestring = (LineString) Functions.makeLine(point1, point2);
+        LineString expectedLineString = GEOMETRY_FACTORY.createLineString(coordArray3d( 1, 1, 1, 2, 2, 2));
+        assertEquals(wktWriter3D.write(acutalLinestring), wktWriter3D.write(expectedLineString));
+
+        MultiPoint multiPoint1 = GEOMETRY_FACTORY.createMultiPointFromCoords(coordArray3d(0.5, 0.5, 1, 1, 1, 1));
+        MultiPoint multiPoint2 = GEOMETRY_FACTORY.createMultiPointFromCoords(coordArray3d(0.5, 0.5, 2, 2, 2, 2));
+        acutalLinestring = (LineString) Functions.makeLine(multiPoint1, multiPoint2);
+        expectedLineString = GEOMETRY_FACTORY.createLineString(coordArray3d( 0.5, 0.5, 1, 1, 1, 1, 0.5, 0.5, 2, 2, 2, 2));
+        assertEquals(wktWriter3D.write(acutalLinestring), wktWriter3D.write(expectedLineString));
+
+        LineString line1 = GEOMETRY_FACTORY.createLineString(coordArray3d(0, 0, 1, 1, 1, 1));
+        LineString line2 = GEOMETRY_FACTORY.createLineString(coordArray3d(2, 2, 2, 2, 3, 3));
+        Geometry[] geoms = new Geometry[]{line1, line2};
+        acutalLinestring = (LineString) Functions.makeLine(geoms);
+        expectedLineString = GEOMETRY_FACTORY.createLineString(coordArray3d( 0, 0, 1, 1, 1, 1, 2, 2, 2, 2, 3, 3));
+        assertEquals(wktWriter3D.write(acutalLinestring), wktWriter3D.write(expectedLineString));
+    }
+
+    @Test
+    public void makeLineWithWrongType() {
+        Polygon polygon1 = GEOMETRY_FACTORY.createPolygon(coordArray(0, 0, 0, 90, 0, 0));
+        Polygon polygon2 = GEOMETRY_FACTORY.createPolygon(coordArray(0, 0, 0, 10, 10, 10, 10, 0, 0, 0));
+
+        Exception e = assertThrows(IllegalArgumentException.class, () -> Functions.makeLine(polygon1, polygon2));
+        assertEquals("ST_MakeLine only supports Point, MultiPoint and LineString geometries", e.getMessage());
+    }
+
     @Test
     public void minimumBoundingRadius() {
         Point point = GEOMETRY_FACTORY.createPoint(new Coordinate(0, 0));

docs/api/flink/Function.md

@@ -1500,6 +1500,40 @@ Output:
 LINESTRING (69.28469348539744 94.28469348539744, 100 125, 111.70035626068274 140.21046313888758)
 ```
 
+## ST_MakeLine
+
+Introduction: Creates a LineString containing the points of Point, MultiPoint, or LineString geometries. Other geometry types cause an error.
+
+Format: `ST_MakeLine(geom1: geometry, geom2: geometry)`
+
+Format: `ST_MakeLine(geoms: array<geometry>)`
+
+Since: `v1.5.0`
+
+Example:
+
+```sql
+SELECT ST_AsText( ST_MakeLine(ST_Point(1,2), ST_Point(3,4)) );
+```
+
+Output:
+
+```
+LINESTRING(1 2,3 4)
+```
+
+Example:
+
+```sql
+SELECT ST_AsText( ST_MakeLine( 'LINESTRING(0 0, 1 1)', 'LINESTRING(2 2, 3 3)' ) );
+```
+
+Output:
+
+```
+ LINESTRING(0 0,1 1,2 2,3 3)
+```
+
 ## ST_MakePolygon
 
 Introduction: Function to convert closed linestring to polygon including holes

docs/api/sql/Function.md

@@ -1508,6 +1508,40 @@ Output:
 LINESTRING (69.28469348539744 94.28469348539744, 100 125, 111.70035626068274 140.21046313888758)
 ```
 
+## ST_MakeLine
+
+Introduction: Creates a LineString containing the points of Point, MultiPoint, or LineString geometries. Other geometry types cause an error.
+
+Format: `ST_MakeLine(geom1: geometry, geom2: geometry)`
+
+Format: `ST_MakeLine(geoms: array<geometry>)`
+
+Since: `v1.5.0`
+
+Example:
+
+```sql
+SELECT ST_AsText( ST_MakeLine(ST_Point(1,2), ST_Point(3,4)) );
+```
+
+Output:
+
+```
+LINESTRING(1 2,3 4)
+```
+
+Example:
+
+```sql
+SELECT ST_AsText( ST_MakeLine( 'LINESTRING(0 0, 1 1)', 'LINESTRING(2 2, 3 3)' ) );
+```
+
+Output:
+
+```
+ LINESTRING(0 0,1 1,2 2,3 3)
+```
+
 ## ST_MakePolygon
 
 Introduction: Function to convert closed linestring to polygon including holes

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

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

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

@@ -686,6 +686,22 @@ public Geometry eval(@DataTypeHint(value = "RAW", bridgedTo = org.locationtech.j
             return org.apache.sedona.common.Functions.lineSubString(geom, startFraction, endFraction);
         }
     }
+
+    public static class ST_MakeLine 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) {
+            Geometry geom1 = (Geometry) o1;
+            Geometry geom2 = (Geometry) o2;
+            return org.apache.sedona.common.Functions.makeLine(geom1, geom2);
+        }
+
+        @DataTypeHint(value = "RAW", bridgedTo = org.locationtech.jts.geom.Geometry.class)
+        public Geometry eval(@DataTypeHint(inputGroup = InputGroup.ANY) Object o) {
+            Geometry[] geoms = (Geometry[]) o;
+            return org.apache.sedona.common.Functions.makeLine(geoms);
+        }
+    }
 
     public static class ST_MakePolygon extends ScalarFunction {
         @DataTypeHint(value = "RAW", bridgedTo = org.locationtech.jts.geom.Geometry.class)

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

@@ -786,6 +786,14 @@ public void testLineSubString() {
         assertEquals("LINESTRING (1 0, 2 0)", result.toString());
     }
 
+    @Test
+    public void testMakeLine() {
+        Table table = tableEnv.sqlQuery("SELECT ST_GeomFromWKT('POINT (0 0)') AS point1, ST_GeomFromWKT('POINT (1 1)') AS point2");
+        table = table.select(call(Functions.ST_MakeLine.class.getSimpleName(), $("point1"), $("point2")));
+        Geometry result = (Geometry) first(table).getField(0);
+        assertEquals("LINESTRING (0 0, 1 1)", result.toString());
+    }
+
     @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

@@ -75,6 +75,7 @@
     "ST_LineInterpolatePoint",
     "ST_LineMerge",
     "ST_LineSubstring",
+    "ST_MakeLine",
     "ST_MakePolygon",
     "ST_MakeValid",
     "ST_MinimumBoundingCircle",
@@ -812,6 +813,20 @@ def ST_LineSubstring(line_string: ColumnOrName, start_fraction: ColumnOrNameOrNu
     return _call_st_function("ST_LineSubstring", (line_string, start_fraction, end_fraction))
 
 
+@validate_argument_types
+def ST_MakeLine(geom1: ColumnOrName, geom2: Optional[ColumnOrName] = None) -> Column:
+    """Creates a LineString containing the points of Point, MultiPoint, or LineString geometries. Other geometry types cause an error.
+
+    :param geom1: Geometry column to convert.
+    :type geometry: ColumnOrName
+    :param geom2: Geometry column to convert. If geoms is empty, then geom1 is an array of geometries.
+    :type geometry: Optional[ColumnOrName], optional
+    :return: LineString geometry column created from the input geomtries.
+    :rtype: Column
+    """
+    args = (geom1,) if geom2 is None else (geom1, geom2)
+    return _call_st_function("ST_MakeLine", args)
+
 @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

@@ -118,6 +118,7 @@
     (stf.ST_LineMerge, ("geom",), "multiline_geom", "", "LINESTRING (0 0, 1 0, 1 1, 0 0)"),
     (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_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

@@ -809,6 +809,26 @@ def test_st_subdivide_explode_lateral(self):
         # Then
         assert lateral_view_result.count() == 16
 
+    def test_st_make_line(self):
+        # Given
+        geometry_df = self.spark.createDataFrame(
+            [
+                ["POINT(0 0)", "POINT(1 1)" , "LINESTRING (0 0, 1 1)"],
+                ["MULTIPOINT ((0 0), (1 1))", "MULTIPOINT ((2 2), (2 3))", "LINESTRING (0 0, 1 1, 2 2, 2 3)"],
+                ["LINESTRING (0 0, 1 1)", "LINESTRING(2 2, 3 3)", "LINESTRING (0 0, 1 1, 2 2, 3 3)"]
+            ]
+        ).selectExpr("ST_GeomFromText(_1) AS geom1", "ST_GeomFromText(_2) AS geom2", "_3 AS expected")
+
+        # When calling st_MakeLine
+        geom_lines = geometry_df.withColumn("linestring", expr("ST_MakeLine(geom1, geom2)"))
+
+        # Then
+        result = geom_lines.selectExpr("ST_AsText(linestring)", "expected"). \
+            collect()
+
+        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

@@ -123,6 +123,7 @@ object Catalog {
     function[ST_LineInterpolatePoint](),
     function[ST_SubDivideExplode](),
     function[ST_SubDivide](),
+    function[ST_MakeLine](),
     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

@@ -688,6 +688,14 @@ case class ST_SubDivideExplode(children: Seq[Expression])
   }
 }
 
+case class ST_MakeLine(inputExpressions: Seq[Expression])
+  extends InferredExpression(InferrableFunction.allowRightNull(Functions.makeLine _)) {
+
+  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

@@ -168,6 +168,11 @@ object st_functions extends DataFrameAPI {
   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 = wrapExpression[ST_LineSubstring](lineString, startFraction, endFraction)
 
+  def ST_MakeLine(geoms: Column): Column = wrapExpression[ST_MakeLine](geoms)
+  def ST_MakeLine(geoms: String): Column = wrapExpression[ST_MakeLine](geoms)
+  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_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

@@ -301,6 +301,14 @@ class dataFrameAPITestScala extends TestBaseScala {
       assert(actualResult)
     }
 
+    it("Passed ST_MakeLine") {
+      val pointDf = sparkSession.sql("SELECT ST_Point(0.0, 0.0) AS a, ST_Point(1.0, 1.0) AS b")
+      val df = pointDf.select(ST_MakeLine("a", "b"))
+      val actualResult = df.take(1)(0).get(0).asInstanceOf[Geometry].toText()
+      val expectedResult = "LINESTRING (0 0, 1 1)"
+      assert(actualResult == expectedResult)
+    }
+
     it("Passed ST_MakeValid On Invalid Polygon") {
       val invalidDf = 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 geom")
       val df = invalidDf.select(ST_MakeValid("geom"))

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

@@ -329,6 +329,14 @@ class functionTestScala extends TestBaseScala with Matchers with GeometrySample
       assert(!testtable.take(1)(0).get(1).asInstanceOf[Boolean])
     }
 
+    it("Passed ST_MakeLine") {
+
+      var testtable = sparkSession.sql(
+        "SELECT ST_MakeLine(ST_GeomFromText('POINT(1 2)'), ST_GeomFromText('POINT(3 4)'))"
+      )
+      assert(testtable.take(1)(0).get(0).asInstanceOf[Geometry].toText.equals("LINESTRING (1 2, 3 4)"))
+    }
+
     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")