diff --git a/common/src/main/java/org/apache/sedona/common/Functions.java b/common/src/main/java/org/apache/sedona/common/Functions.java
index be32113fd2..2bed5fbc44 100644
--- a/common/src/main/java/org/apache/sedona/common/Functions.java
+++ b/common/src/main/java/org/apache/sedona/common/Functions.java
@@ -970,8 +970,6 @@ public static Geometry boundingDiagonal(Geometry geometry) {
         if (geometry.isEmpty()) {
             return GEOMETRY_FACTORY.createLineString();
         }else {
-            //Envelope envelope = geometry.getEnvelopeInternal();
-           // if (envelope.isNull()) return GEOMETRY_FACTORY.createLineString();
             Double startX = null, startY = null, startZ = null,
                     endX = null, endY = null, endZ = null;
             boolean is3d = !Double.isNaN(geometry.getCoordinate().z);
@@ -1000,6 +998,34 @@ public static Geometry boundingDiagonal(Geometry geometry) {
         }
     }
 
+    public static double angle(Geometry point1, Geometry point2, Geometry point3, Geometry point4) throws IllegalArgumentException {
+        if (point3 == null && point4 == null) return Functions.angle(point1, point2);
+        else if (point4 == null) return Functions.angle(point1, point2, point3);
+        if (GeomUtils.isAnyGeomEmpty(point1, point2, point3, point4)) throw new IllegalArgumentException("ST_Angle cannot support empty geometries.");
+        if (!(point1 instanceof Point && point2 instanceof Point && point3 instanceof Point && point4 instanceof Point)) throw new IllegalArgumentException("ST_Angle supports either only POINT or only LINESTRING geometries.");
+        return GeomUtils.calcAngle(point1.getCoordinate(), point2.getCoordinate(), point3.getCoordinate(), point4.getCoordinate());
+    }
+
+    public static double angle(Geometry point1, Geometry point2, Geometry point3) throws IllegalArgumentException {
+        if (GeomUtils.isAnyGeomEmpty(point1, point2, point3)) throw new IllegalArgumentException("ST_Angle cannot support empty geometries.");
+        if (!(point1 instanceof Point && point2 instanceof Point && point3 instanceof Point)) throw new IllegalArgumentException("ST_Angle supports either only POINT or only LINESTRING geometries.");
+        return GeomUtils.calcAngle(point2.getCoordinate(), point1.getCoordinate(), point2.getCoordinate(), point3.getCoordinate());
+    }
+
+    public static double angle(Geometry line1, Geometry line2) throws IllegalArgumentException {
+        if (GeomUtils.isAnyGeomEmpty(line1, line2)) throw new IllegalArgumentException("ST_Angle cannot support empty geometries.");
+        if (!(line1 instanceof LineString && line2 instanceof LineString)) throw new IllegalArgumentException("ST_Angle supports either only POINT or only LINESTRING geometries.");
+        Coordinate[] startEndLine1 = GeomUtils.getStartEndCoordinates(line1);
+        Coordinate[] startEndLine2 = GeomUtils.getStartEndCoordinates(line2);
+        assert startEndLine1 != null;
+        assert startEndLine2 != null;
+        return GeomUtils.calcAngle(startEndLine1[0], startEndLine1[1], startEndLine2[0], startEndLine2[1]);
+    }
+
+    public static double degrees(double angleInRadian) {
+        return GeomUtils.toDegrees(angleInRadian);
+    }
+
     public static Double hausdorffDistance(Geometry g1, Geometry g2, double densityFrac) throws Exception {
         return GeomUtils.getHausdorffDistance(g1, g2, densityFrac);
     }
diff --git a/common/src/main/java/org/apache/sedona/common/utils/GeomUtils.java b/common/src/main/java/org/apache/sedona/common/utils/GeomUtils.java
index 2960c779d1..0b0b2e8615 100644
--- a/common/src/main/java/org/apache/sedona/common/utils/GeomUtils.java
+++ b/common/src/main/java/org/apache/sedona/common/utils/GeomUtils.java
@@ -20,6 +20,7 @@
 import org.locationtech.jts.io.WKTWriter;
 import org.locationtech.jts.operation.polygonize.Polygonizer;
 import org.locationtech.jts.operation.union.UnaryUnionOp;
+import org.locationtech.jts.algorithm.Angle;
 import org.locationtech.jts.algorithm.distance.DiscreteFrechetDistance;
 import org.locationtech.jts.algorithm.distance.DiscreteHausdorffDistance;
 
@@ -454,8 +455,43 @@ public static void translateGeom(Geometry geometry, double deltaX, double deltaY
             geometry.geometryChanged();
         }
     }
+
+    public static boolean isAnyGeomEmpty(Geometry... geometries) {
+        for (Geometry geometry: geometries) {
+            if (geometry != null)
+                if (geometry.isEmpty())
+                    return true;
+        }
+        return false;
+    }
+
+    public static Coordinate[] getStartEndCoordinates(Geometry line) {
+        if (line.getNumPoints() < 2) return null;
+        Coordinate[] coordinates = line.getCoordinates();
+        return new Coordinate[] {coordinates[0], coordinates[coordinates.length - 1]};
+    }
+
+    public static double calcAngle(Coordinate start1, Coordinate end1, Coordinate start2, Coordinate end2) {
+        double angle1 = normalizeAngle(Angle.angle(start1, end1));
+        double angle2 = normalizeAngle(Angle.angle(start2, end2));
+        return normalizeAngle(angle1 - angle2);
+    }
+
+    private static double normalizeAngle(double angle) {
+        if (angle < 0) {
+            return 2 * Math.PI - Math.abs(angle);
+        }
+        return angle;
+    }
+
+    public static double toDegrees(double angleInRadian) {
+        return Angle.toDegrees(angleInRadian);
+    }
+
+
     public static void affineGeom(Geometry geometry, Double a, Double b, Double c, Double d, Double e, Double f, Double g, Double h, Double i, Double xOff, Double yOff,
                                   Double zOff) {
+
         Coordinate[] coordinates = geometry.getCoordinates();
         for (Coordinate currCoordinate : coordinates) {
             double x = currCoordinate.getX(), y = currCoordinate.getY(), z = Double.isNaN(currCoordinate.getZ()) ? 0 : currCoordinate.getZ();
diff --git a/common/src/test/java/org/apache/sedona/common/FunctionsTest.java b/common/src/test/java/org/apache/sedona/common/FunctionsTest.java
index 5f65652337..46db35e7ff 100644
--- a/common/src/test/java/org/apache/sedona/common/FunctionsTest.java
+++ b/common/src/test/java/org/apache/sedona/common/FunctionsTest.java
@@ -1033,6 +1033,113 @@ public void boundingDiagonalSingleVertex() {
     }
 
     @Test
+    public void angleFourPoints() {
+        Point start1 = GEOMETRY_FACTORY.createPoint(new Coordinate(0, 0));
+        Point end1 = GEOMETRY_FACTORY.createPoint(new Coordinate(1, 1));
+        Point start2 = GEOMETRY_FACTORY.createPoint(new Coordinate(1, 0));
+        Point end2 = GEOMETRY_FACTORY.createPoint(new Coordinate(6, 2));
+
+        double expected = 0.4048917862850834;
+        double expectedDegrees = 23.198590513648185;
+        double reverseExpectedDegrees = 336.8014094863518;
+        double reverseExpected = 5.878293520894503;
+
+        double actualPointsFour = Functions.angle(start1, end1, start2, end2);
+        double actualPointsFourDegrees = Functions.degrees(actualPointsFour);
+        double actualPointsFourReverse = Functions.angle(start2, end2, start1, end1);
+        double actualPointsFourReverseDegrees = Functions.degrees(actualPointsFourReverse);
+
+        assertEquals(expected, actualPointsFour, 1e-9);
+        assertEquals(expectedDegrees, actualPointsFourDegrees, 1e-9);
+        assertEquals(reverseExpected, actualPointsFourReverse, 1e-9);
+        assertEquals(reverseExpectedDegrees, actualPointsFourReverseDegrees, 1e-9);
+    }
+
+    @Test
+    public void angleFourPoints3D() {
+        Point start1 = GEOMETRY_FACTORY.createPoint(new Coordinate(0, 0, 4));
+        Point end1 = GEOMETRY_FACTORY.createPoint(new Coordinate(1, 1, 5));
+        Point start2 = GEOMETRY_FACTORY.createPoint(new Coordinate(1, 0, 9));
+        Point end2 = GEOMETRY_FACTORY.createPoint(new Coordinate(6, 2, 2));
+
+        double expected = 0.4048917862850834;
+        double expectedDegrees = 23.198590513648185;
+        double reverseExpectedDegrees = 336.8014094863518;
+        double reverseExpected = 5.878293520894503;
+
+        double actualPointsFour = Functions.angle(start1, end1, start2, end2);
+        double actualPointsFourDegrees = Functions.degrees(actualPointsFour);
+        double actualPointsFourReverse = Functions.angle(start2, end2, start1, end1);
+        double actualPointsFourReverseDegrees = Functions.degrees(actualPointsFourReverse);
+
+        assertEquals(expected, actualPointsFour, 1e-9);
+        assertEquals(expectedDegrees, actualPointsFourDegrees, 1e-9);
+        assertEquals(reverseExpected, actualPointsFourReverse, 1e-9);
+        assertEquals(reverseExpectedDegrees, actualPointsFourReverseDegrees, 1e-9);
+    }
+
+
+
+    @Test
+    public void angleThreePoints() {
+        Point point1 = GEOMETRY_FACTORY.createPoint(new Coordinate(1, 1));
+        Point point2 = GEOMETRY_FACTORY.createPoint(new Coordinate(0, 0));
+        Point point3 = GEOMETRY_FACTORY.createPoint(new Coordinate(3, 2));
+
+        double expected = 0.19739555984988044;
+        double expectedDegrees = 11.309932474020195;
+
+
+        double actualPointsThree = Functions.angle(point1, point2, point3);
+        double actualPointsThreeDegrees = Functions.degrees(actualPointsThree);
+
+        assertEquals(expected, actualPointsThree, 1e-9);
+        assertEquals(expectedDegrees, actualPointsThreeDegrees, 1e-9);
+
+    }
+
+    @Test
+    public void angleTwoLineStrings() {
+        LineString lineString1 = GEOMETRY_FACTORY.createLineString(coordArray(0, 0, 1, 1));
+        LineString lineString2 = GEOMETRY_FACTORY.createLineString(coordArray(0, 0, 3, 2));
+
+        double expected = 0.19739555984988044;
+        double expectedDegrees = 11.309932474020195;
+        double reverseExpected = 6.085789747329706;
+        double reverseExpectedDegrees = 348.69006752597977;
+
+        double actualLineString = Functions.angle(lineString1, lineString2);
+        double actualLineStringReverse = Functions.angle(lineString2, lineString1);
+        double actualLineStringDegrees = Functions.degrees(actualLineString);
+        double actualLineStringReverseDegrees = Functions.degrees(actualLineStringReverse);
+
+        assertEquals(expected, actualLineString, 1e-9);
+        assertEquals(reverseExpected, actualLineStringReverse, 1e-9);
+        assertEquals(expectedDegrees, actualLineStringDegrees, 1e-9);
+        assertEquals(reverseExpectedDegrees, actualLineStringReverseDegrees, 1e-9);
+    }
+
+    @Test
+    public void angleInvalidEmptyGeom() {
+        Point point1 = GEOMETRY_FACTORY.createPoint(new Coordinate(3, 5));
+        Point point2 = GEOMETRY_FACTORY.createPoint();
+        Point point3 = GEOMETRY_FACTORY.createPoint(new Coordinate(1, 1));
+
+        Exception e = assertThrows(IllegalArgumentException.class, () -> Functions.angle(point1, point2, point3));
+        assertEquals("ST_Angle cannot support empty geometries.", e.getMessage());
+    }
+
+    @Test
+    public void angleInvalidUnsupportedGeom() {
+        Point point1 = GEOMETRY_FACTORY.createPoint(new Coordinate(3, 5));
+        Polygon polygon = GEOMETRY_FACTORY.createPolygon(coordArray(1, 0, 1, 1, 2, 1, 2, 0, 1, 0));
+        Point point3 = GEOMETRY_FACTORY.createPoint(new Coordinate(1, 1));
+
+        Exception e = assertThrows(IllegalArgumentException.class, () -> Functions.angle(point1, polygon, point3));
+        assertEquals("ST_Angle supports either only POINT or only LINESTRING geometries.", e.getMessage());
+    }
+
+
     public void affineEmpty3D() {
         LineString emptyLineString = GEOMETRY_FACTORY.createLineString();
         String expected = emptyLineString.toText();
diff --git a/docs/api/flink/Function.md b/docs/api/flink/Function.md
index fa7f7f1481..0ed05b6caf 100644
--- a/docs/api/flink/Function.md
+++ b/docs/api/flink/Function.md
@@ -131,6 +131,74 @@ Input: `POLYGON ((1 0 1, 1 1 1, 2 2 2, 1 0 1))`
 
 Output: `POLYGON Z((2 3 1, 4 5 1, 7 8 2, 2 3 1))`
 
+## ST_Angle
+
+Introduction: Compute and return the angle between two vectors represented by the provided points or linestrings.
+
+There are three variants possible for ST_Angle:
+
+`ST_Angle(Geometry point1, Geometry point2, Geometry point3, Geometry point4)`
+Computes the angle formed by vectors represented by point1 - point2 and point3 - point4
+
+`ST_Angle(Geometry point1, Geometry point2, Geometry point3)`
+Computes the angle formed by vectors represented by point2 - point1 and point2 - point3
+
+`ST_Angle(Geometry line1, Geometry line2)`
+Computes the angle formed by vectors S1 - E1 and S2 - E2, where S and E denote start and end points respectively
+
+!!!Note
+    If any other geometry type is provided, ST_Angle throws an IllegalArgumentException.
+    Additionally, if any of the provided geometry is empty, ST_Angle throws an IllegalArgumentException.
+
+!!!Note
+    If a 3D geometry is provided, ST_Angle computes the angle ignoring the z ordinate, equivalent to calling ST_Angle for corresponding 2D geometries.
+
+!!!Tip
+    ST_Angle returns the angle in radian between 0 and 2\Pi. To convert the angle to degrees, use [ST_Degrees](./#st_degrees).
+
+
+Format: `ST_Angle(p1, p2, p3, p4) | ST_Angle(p1, p2, p3) | ST_Angle(line1, line2)`
+
+Since: `1.5.0`
+
+Example:
+
+```sql
+ST_Angle(p1, p2, p3, p4)
+```
+
+Input: `p1: POINT (0 0)`
+
+Input: `p2: POINT (1 1)`
+
+Input: `p3: POINT (1 0)`
+
+Input: `p4: POINT(6 2)`
+
+Output: 0.4048917862850834
+
+```sql
+ST_Angle(p1, p2, p3)
+```
+
+Input: `p1: POINT (1 1)`
+
+Input: `p2: POINT (0 0)`
+
+Input: `p3: POINT(3 2)`
+
+Output: 0.19739555984988044
+
+```sql
+ST_Angle(line1, line2)
+```
+
+Input: `line1: LINESTRING (0 0, 1 1)`
+
+Input: `line2: LINESTRING (0 0, 3 2)`
+
+Output: 0.19739555984988044
+
 ## ST_Area
 
 Introduction: Return the area of A
@@ -462,6 +530,22 @@ SELECT ST_DistanceSpheroid(ST_GeomFromWKT('POINT (51.3168 -0.56)'), ST_GeomFromW
 
 Output: `544430.9411996207`
 
+## ST_Degrees
+
+Introduction: Convert an angle in radian to degrees.
+
+Format: `ST_Degrees(angleInRadian)`
+
+Since: `1.5.0`
+
+Example:
+
+```sql
+SELECT ST_Degrees(0.19739555984988044)
+```
+
+Output: 11.309932474020195
+
 ## ST_Envelope
 
 Introduction: Return the envelop boundary of A
diff --git a/docs/api/sql/Function.md b/docs/api/sql/Function.md
index a570500f07..698f342579 100644
--- a/docs/api/sql/Function.md
+++ b/docs/api/sql/Function.md
@@ -93,10 +93,6 @@ If the given geometry is empty, the result is also empty.
 Format: `ST_Affine(geometry, a, b, c, d, e, f, g, h, i, xOff, yOff, zOff)`  
 Format: `ST_Affine(geometry, a, b, d, e, xOff, yOff)`
 
-Since: `1.5.0`
-
-Example:
-
 ```sql
 ST_Affine(geometry, 1, 2, 4, 1, 1, 2, 3, 2, 5, 4, 8, 3)
 ```
@@ -130,6 +126,74 @@ Input: `POLYGON ((1 0 1, 1 1 1, 2 2 2, 1 0 1))`
 
 Output: `POLYGON Z((2 3 1, 4 5 1, 7 8 2, 2 3 1))`
 
+## ST_Angle
+
+Introduction: Computes and returns the angle between two vectors represented by the provided points or linestrings.
+
+There are three variants possible for ST_Angle:
+
+`ST_Angle(Geometry point1, Geometry point2, Geometry point3, Geometry point4)`
+Computes the angle formed by vectors represented by point1 - point2 and point3 - point4
+
+`ST_Angle(Geometry point1, Geometry point2, Geometry point3)`
+Computes the angle formed by vectors represented by point2 - point1 and point2 - point3
+
+`ST_Angle(Geometry line1, Geometry line2)`
+Computes the angle formed by vectors S1 - E1 and S2 - E2, where S and E denote start and end points respectively
+
+!!!Note
+    If any other geometry type is provided, ST_Angle throws an IllegalArgumentException.
+    Additionally, if any of the provided geometry is empty, ST_Angle throws an IllegalArgumentException.
+
+!!!Note
+    If a 3D geometry is provided, ST_Angle computes the angle ignoring the z ordinate, equivalent to calling ST_Angle for corresponding 2D geometries.
+
+!!!Tip
+    ST_Angle returns the angle in radian between 0 and 2\Pi. To convert the angle to degrees, use [ST_Degrees](./#st_degrees).
+
+
+Format: `ST_Angle(p1, p2, p3, p4) | ST_Angle(p1, p2, p3) | ST_Angle(line1, line2)`
+
+
+Since: `1.5.0`
+
+Example:
+
+```sql
+ST_Angle(p1, p2, p3, p4)
+```
+
+Input: `p1: POINT (0 0)`
+
+Input: `p2: POINT (1 1)`
+
+Input: `p3: POINT (1 0)`
+
+Input: `p4: POINT(6 2)`
+
+Output: 0.4048917862850834
+
+```sql
+ST_Angle(p1, p2, p3)
+```
+
+Input: `p1: POINT (1 1)`
+
+Input: `p2: POINT (0 0)`
+
+Input: `p3: POINT(3 2)`
+
+Output: 0.19739555984988044
+
+```sql
+ST_Angle(line1, line2)
+```
+
+Input: `line1: LINESTRING (0 0, 1 1)`
+
+Input: `line2: LINESTRING (0 0, 3 2)`
+
+Output: 0.19739555984988044
 
 ## ST_Area
 
@@ -503,6 +567,22 @@ SELECT ST_ConvexHull(polygondf.countyshape)
 FROM polygondf
 ```
 
+## ST_Degrees
+
+Introduction: Convert an angle in radian to degrees.
+
+Format: `ST_Degrees(angleInRadian)`
+
+Since: `1.5.0`
+
+Example:
+
+```sql
+SELECT ST_Degrees(0.19739555984988044)
+```
+
+Output: 11.309932474020195
+
 ## ST_Difference
 
 Introduction: Return the difference between geometry A and B (return part of geometry A that does not intersect geometry B)
diff --git a/flink/src/main/java/org/apache/sedona/flink/Catalog.java b/flink/src/main/java/org/apache/sedona/flink/Catalog.java
index 2e515dbdfb..eb455c3d57 100644
--- a/flink/src/main/java/org/apache/sedona/flink/Catalog.java
+++ b/flink/src/main/java/org/apache/sedona/flink/Catalog.java
@@ -104,6 +104,8 @@ public static UserDefinedFunction[] getFuncs() {
                 new Functions.ST_FrechetDistance(),
                 new Functions.ST_Affine(),
                 new Functions.ST_BoundingDiagonal(),
+                new Functions.ST_Angle(),
+                new Functions.ST_Degrees(),
                 new Functions.ST_HausdorffDistance(),
         };
     }
diff --git a/flink/src/main/java/org/apache/sedona/flink/expressions/Functions.java b/flink/src/main/java/org/apache/sedona/flink/expressions/Functions.java
index 576d668bb8..b57a559648 100644
--- a/flink/src/main/java/org/apache/sedona/flink/expressions/Functions.java
+++ b/flink/src/main/java/org/apache/sedona/flink/expressions/Functions.java
@@ -693,4 +693,47 @@ public Double eval(@DataTypeHint(value = "RAW", bridgedTo = org.locationtech.jts
             return org.apache.sedona.common.Functions.hausdorffDistance(geom1, geom2);
         }
     }
+
+    public static class ST_Angle extends ScalarFunction {
+
+        @DataTypeHint("Double")
+        public Double eval(@DataTypeHint(value = "RAW", bridgedTo = org.locationtech.jts.geom.Geometry.class) Object p1,
+                           @DataTypeHint(value = "RAW", bridgedTo = org.locationtech.jts.geom.Geometry.class) Object p2,
+                           @DataTypeHint(value = "RAW", bridgedTo = org.locationtech.jts.geom.Geometry.class) Object p3,
+                           @DataTypeHint(value = "RAW", bridgedTo = org.locationtech.jts.geom.Geometry.class) Object p4) {
+            Geometry point1 = (Geometry) p1;
+            Geometry point2 = (Geometry) p2;
+            Geometry point3 = (Geometry) p3;
+            Geometry point4 = (Geometry) p4;
+
+            return org.apache.sedona.common.Functions.angle(point1, point2, point3, point4);
+        }
+
+        @DataTypeHint("Double")
+        public Double eval(@DataTypeHint(value = "RAW", bridgedTo = org.locationtech.jts.geom.Geometry.class) Object p1,
+                           @DataTypeHint(value = "RAW", bridgedTo = org.locationtech.jts.geom.Geometry.class) Object p2,
+                           @DataTypeHint(value = "RAW", bridgedTo = org.locationtech.jts.geom.Geometry.class) Object p3) {
+            Geometry point1 = (Geometry) p1;
+            Geometry point2 = (Geometry) p2;
+            Geometry point3 = (Geometry) p3;
+
+            return org.apache.sedona.common.Functions.angle(point1, point2, point3);
+        }
+
+        @DataTypeHint("Double")
+        public Double eval(@DataTypeHint(value = "RAW", bridgedTo = org.locationtech.jts.geom.Geometry.class) Object line1,
+                           @DataTypeHint(value = "RAW", bridgedTo = org.locationtech.jts.geom.Geometry.class) Object line2) {
+            Geometry lineString1 = (Geometry) line1;
+            Geometry lineString2 = (Geometry) line2;
+
+            return org.apache.sedona.common.Functions.angle(lineString1, lineString2);
+        }
+    }
+
+    public static class ST_Degrees extends ScalarFunction {
+        @DataTypeHint("Double")
+        public Double eval(@DataTypeHint("Double") Double angleInRadian) {
+            return org.apache.sedona.common.Functions.degrees(angleInRadian);
+        }
+    }
 }
diff --git a/flink/src/test/java/org/apache/sedona/flink/FunctionTest.java b/flink/src/test/java/org/apache/sedona/flink/FunctionTest.java
index 22bb602b90..8ed4916c87 100644
--- a/flink/src/test/java/org/apache/sedona/flink/FunctionTest.java
+++ b/flink/src/test/java/org/apache/sedona/flink/FunctionTest.java
@@ -788,6 +788,16 @@ public void testBoundingDiagonal() {
         assertEquals(expected, actual);
     }
 
+    @Test
+    public void testAngle() {
+        Table polyTable = tableEnv.sqlQuery("SELECT ST_Angle(ST_GeomFromWKT('LINESTRING (0 0, 1 1)'), ST_GeomFromWKT('LINESTRING (0 0, 3 2)'))" + " AS " + polygonColNames[0]);
+        polyTable = polyTable.select(call(Functions.ST_Degrees.class.getSimpleName(), $(polygonColNames[0])));
+        Double expected = 11.309932474020195;
+        Double actual = (Double) first(polyTable).getField(0);
+        assertEquals(expected, actual, 1e-9);
+
+    }
+
     @Test
     public void testHausdorffDistance() {
         Table polyTable = tableEnv.sqlQuery("SELECT ST_GeomFromWKT('POINT (0.0 1.0)') AS g1, ST_GeomFromWKT('LINESTRING (0 0, 1 0, 2 0, 3 0, 4 0, 5 0)') AS g2");
diff --git a/python/sedona/sql/st_functions.py b/python/sedona/sql/st_functions.py
index 47da57ffa5..1cd944cb07 100644
--- a/python/sedona/sql/st_functions.py
+++ b/python/sedona/sql/st_functions.py
@@ -114,6 +114,8 @@
     "ST_Force3D",
     "ST_NRings",
     "ST_Translate",
+    "ST_Angle",
+    "ST_Degrees",
     "ST_FrechetDistance",
     "ST_Affine",
     "ST_BoundingDiagonal"
@@ -1356,6 +1358,41 @@ def ST_BoundingDiagonal(geometry: ColumnOrName) -> Column:
 
     return _call_st_function("ST_BoundingDiagonal", geometry)
 
+
+@validate_argument_types
+def ST_Angle(g1: ColumnOrName, g2: ColumnOrName, g3: Optional[ColumnOrName] = None, g4: Optional[ColumnOrName] = None) -> Column:
+    """
+    Returns the computed angle between vectors formed by given geometries in radian. Range of result is between 0 and 2 * pi.
+    3 Variants:
+        Angle(Point1, Point2, Point3, Point4)
+            Computes angle formed by vectors formed by Point1-Point2 and Point3-Point4
+        Angle(Point1, Point2, Point3)
+            Computes angle formed by angle Point1-Point2-Point3
+        Angle(Line1, Line2)
+            Computes angle between vectors formed by S1-E1 and S2-E2, where S and E are start and endpoints.
+    :param g1: Point or Line
+    :param g2: Point or Line
+    :param g3: Point or None
+    :param g4: Point or None
+    :return: Returns the computed angle
+    """
+    args = (g1, g2)
+    if g3 is not None:
+        if g4 is not None:
+            args = (g1, g2, g3, g4)
+        else:
+            args = (g1, g2, g3)
+    # args = (g1, g2, g3, g4)
+    return _call_st_function("ST_Angle", args)
+
+@validate_argument_types
+def ST_Degrees(angleInRadian: Union[ColumnOrName, float]) -> Column:
+    """
+    Converts a given angle from radian to degrees
+    :param angleInRadian: Angle in Radian
+    :return: Angle in Degrees
+    """
+    return _call_st_function("ST_Degrees", angleInRadian)
 @validate_argument_types
 def ST_HausdorffDistance(g1: ColumnOrName, g2: ColumnOrName, densityFrac: Optional[Union[ColumnOrName, float]] = -1) -> Column:
     """
diff --git a/python/tests/sql/test_dataframe_api.py b/python/tests/sql/test_dataframe_api.py
index 4ab13edac1..5337604a6c 100644
--- a/python/tests/sql/test_dataframe_api.py
+++ b/python/tests/sql/test_dataframe_api.py
@@ -55,6 +55,10 @@
     (stf.ST_Affine, ("geom", 1.0, 2.0, 1.0, 2.0, 1.0, 2.0,), "square_geom", "", "POLYGON ((2 3, 4 5, 5 6, 3 4, 2 3))"),
     (stf.ST_AddPoint, ("line", lambda: f.expr("ST_Point(1.0, 1.0)")), "linestring_geom", "", "LINESTRING (0 0, 1 0, 2 0, 3 0, 4 0, 5 0, 1 1)"),
     (stf.ST_AddPoint, ("line", lambda: f.expr("ST_Point(1.0, 1.0)"), 1), "linestring_geom", "", "LINESTRING (0 0, 1 1, 1 0, 2 0, 3 0, 4 0, 5 0)"),
+    (stf.ST_Angle, ("p1", "p2", "p3", "p4", ), "four_points", "", 0.4048917862850834),
+    (stf.ST_Angle, ("p1", "p2", "p3",), "three_points", "", 0.19739555984988078),
+    (stf.ST_Angle, ("line1", "line2"), "two_lines", "", 0.19739555984988078),
+    (stf.ST_Degrees, ("angleRad",), "two_lines_angle_rad", "", 11.309932474020213),
     (stf.ST_Area, ("geom",), "triangle_geom", "", 0.5),
     (stf.ST_AreaSpheroid, ("point",), "point_geom", "", 0.0),
     (stf.ST_AsBinary, ("point",), "point_geom", "", "01010000000000000000000000000000000000f03f"),
@@ -397,6 +401,14 @@ def base_df(self, request):
             return TestDataFrameAPI.spark.sql("SELECT ST_GeomFromWKT('LINESTRING (0 0, 2 1)') AS line, ST_GeomFromWKT('POLYGON ((1 0, 2 0, 2 2, 1 2, 1 0))') AS poly")
         elif request.param == "square_geom":
             return TestDataFrameAPI.spark.sql("SELECT ST_GeomFromWKT('POLYGON ((1 0, 1 1, 2 1, 2 0, 1 0))') AS geom")
+        elif request.param == "four_points":
+            return TestDataFrameAPI.spark.sql("SELECT ST_GeomFromWKT('POINT (0 0)') AS p1, ST_GeomFromWKT('POINT (1 1)') AS p2, ST_GeomFromWKT('POINT (1 0)') AS p3, ST_GeomFromWKT('POINT (6 2)') AS p4")
+        elif request.param == "three_points":
+            return TestDataFrameAPI.spark.sql("SELECT ST_GeomFromWKT('POINT (1 1)') AS p1, ST_GeomFromWKT('POINT (0 0)') AS p2, ST_GeomFromWKT('POINT (3 2)') AS p3")
+        elif request.param == "two_lines":
+            return TestDataFrameAPI.spark.sql("SELECT ST_GeomFromWKT('LINESTRING (0 0, 1 1)') AS line1, ST_GeomFromWKT('LINESTRING (0 0, 3 2)') AS line2")
+        elif request.param == "two_lines_angle_rad":
+            return TestDataFrameAPI.spark.sql("SELECT ST_Angle(ST_GeomFromWKT('LINESTRING (0 0, 1 1)'), ST_GeomFromWKT('LINESTRING (0 0, 3 2)')) AS angleRad")
         elif request.param == "geometry_geom_collection":
             return TestDataFrameAPI.spark.sql("SELECT ST_GeomFromWKT('GEOMETRYCOLLECTION(POINT(1 1), LINESTRING(0 0, 1 1, 2 2))') AS geom")
         elif request.param == "point_and_line":
diff --git a/python/tests/sql/test_function.py b/python/tests/sql/test_function.py
index 7c2967fdd8..13137beceb 100644
--- a/python/tests/sql/test_function.py
+++ b/python/tests/sql/test_function.py
@@ -1149,6 +1149,14 @@ def test_boundingDiagonal(self):
         actual = actual_df.selectExpr("ST_AsText(geom)").take(1)[0][0]
         assert expected == actual
 
+    def test_angle(self):
+        expectedDegrees = 11.309932474020195
+        expectedRad = 0.19739555984988044
+        actual_df = self.spark.sql("SELECT ST_Angle(ST_GeomFromText('LINESTRING (0 0, 1 1)'), ST_GeomFromText('LINESTRING (0 0, 3 2)')) AS angleRad")
+        actualRad = actual_df.take(1)[0][0]
+        actualDegrees = actual_df.selectExpr("ST_Degrees(angleRad)").take(1)[0][0]
+        assert math.isclose(expectedRad, actualRad, rel_tol=1e-9)
+        assert math.isclose(expectedDegrees, actualDegrees, rel_tol=1e-9)
     def test_hausdorffDistance(self):
         expected = 5.0
         actual_df = self.spark.sql("SELECT ST_HausdorffDistance(ST_GeomFromText('POLYGON ((1 0 1, 1 1 2, 2 1 5, "
diff --git a/sql/common/src/main/scala/org/apache/sedona/sql/UDF/Catalog.scala b/sql/common/src/main/scala/org/apache/sedona/sql/UDF/Catalog.scala
index 096fee0c85..df6134b308 100644
--- a/sql/common/src/main/scala/org/apache/sedona/sql/UDF/Catalog.scala
+++ b/sql/common/src/main/scala/org/apache/sedona/sql/UDF/Catalog.scala
@@ -156,6 +156,8 @@ object Catalog {
     function[ST_FrechetDistance](),
     function[ST_Affine](),
     function[ST_BoundingDiagonal](),
+    function[ST_Angle](),
+    function[ST_Degrees](),
     function[ST_HausdorffDistance](-1),
     // Expression for rasters
     function[RS_NormalizedDifference](),
diff --git a/sql/common/src/main/scala/org/apache/spark/sql/sedona_sql/expressions/Functions.scala b/sql/common/src/main/scala/org/apache/spark/sql/sedona_sql/expressions/Functions.scala
index 7b79175d1e..c661fc92a4 100644
--- a/sql/common/src/main/scala/org/apache/spark/sql/sedona_sql/expressions/Functions.scala
+++ b/sql/common/src/main/scala/org/apache/spark/sql/sedona_sql/expressions/Functions.scala
@@ -1042,9 +1042,24 @@ case class ST_HausdorffDistance(inputExpressions: Seq[Expression])
   }
 }
 
+case class ST_Angle(inputExpressions: Seq[Expression])
+  extends InferredExpression(inferrableFunction4(Functions.angle _), inferrableFunction3(Functions.angle _), inferrableFunction2(Functions.angle _)) with FoldableExpression {
+  protected def withNewChildrenInternal(newChildren: IndexedSeq[Expression]) = {
+    copy(inputExpressions = newChildren)
+  }
+}
+
 case class GeometryType(inputExpressions: Seq[Expression])
   extends InferredExpression(Functions.geometryTypeWithMeasured _) with FoldableExpression {
   protected def withNewChildrenInternal(newChildren: IndexedSeq[Expression]) = {
     copy(inputExpressions = newChildren)
   }
 }
+
+case class ST_Degrees(inputExpressions: Seq[Expression])
+  extends InferredExpression(Functions.degrees _) with FoldableExpression {
+  protected def withNewChildrenInternal(newChildren: IndexedSeq[Expression]) = {
+    copy(inputExpressions = newChildren)
+  }
+}
+
diff --git a/sql/common/src/main/scala/org/apache/spark/sql/sedona_sql/expressions/InferredExpression.scala b/sql/common/src/main/scala/org/apache/spark/sql/sedona_sql/expressions/InferredExpression.scala
index fd29123d38..3d8ade3b75 100644
--- a/sql/common/src/main/scala/org/apache/spark/sql/sedona_sql/expressions/InferredExpression.scala
+++ b/sql/common/src/main/scala/org/apache/spark/sql/sedona_sql/expressions/InferredExpression.scala
@@ -235,44 +235,4 @@ object InferrableFunction {
     })
   }
 
-  def allowSixRightNull[R, A1, A2, A3, A4, A5, A6, A7, A8, A9, A10, A11, A12, A13](f: (A1, A2, A3, A4, A5, A6, A7, A8, A9, A10, A11, A12, A13) => R)
-                                                                                  (implicit typeTag: TypeTag[(A1, A2, A3, A4, A5, A6, A7, A8, A9, A10, A11, A12, A13) => R]): InferrableFunction = {
-    apply(typeTag, extractors => {
-      val func = f.asInstanceOf[(Any, Any, Any, Any, Any, Any, Any, Any, Any, Any, Any, Any, Any) => Any]
-      val extractor1 = extractors(0)
-      val extractor2 = extractors(1)
-      val extractor3 = extractors(2)
-      val extractor4 = extractors(3)
-      val extractor5 = extractors(4)
-      val extractor6 = extractors(5)
-      val extractor7 = extractors(6)
-      val extractor8 = extractors(7)
-      val extractor9 = extractors(8)
-      val extractor10 = extractors(9)
-      val extractor11 = extractors(10)
-      val extractor12 = extractors(11)
-      val extractor13 = extractors(12)
-      input => {
-        val arg1 = extractor1(input)
-        val arg2 = extractor2(input)
-        val arg3 = extractor3(input)
-        val arg4 = extractor4(input)
-        val arg5 = extractor5(input)
-        val arg6 = extractor6(input)
-        val arg7 = extractor7(input)
-        val arg8 = extractor8(input)
-        val arg9 = extractor9(input)
-        val arg10 = extractor10(input)
-        val arg11 = extractor11(input)
-        val arg12 = extractor12(input)
-        val arg13 = extractor13(input)
-        if (arg1 != null && arg2 != null && arg3 != null && arg4 != null && arg5 != null && arg6 != null && arg7 != null) {
-          func(arg1, arg2, arg3, arg4, arg5, arg6, arg7, arg8, arg9, arg10, arg11, arg12, arg13)
-        } else {
-          null
-        }
-      }
-    })
-  }
-
 }
diff --git a/sql/common/src/main/scala/org/apache/spark/sql/sedona_sql/expressions/st_functions.scala b/sql/common/src/main/scala/org/apache/spark/sql/sedona_sql/expressions/st_functions.scala
index f74e2413bf..210529a9a6 100644
--- a/sql/common/src/main/scala/org/apache/spark/sql/sedona_sql/expressions/st_functions.scala
+++ b/sql/common/src/main/scala/org/apache/spark/sql/sedona_sql/expressions/st_functions.scala
@@ -353,6 +353,21 @@ object st_functions extends DataFrameAPI {
   def ST_BoundingDiagonal(geometry: String) =
     wrapExpression[ST_BoundingDiagonal](geometry)
 
+  def ST_Angle(p1: Column, p2: Column, p3: Column, p4: Column): Column = wrapExpression[ST_Angle](p1, p2, p3, p4)
+
+  def ST_Angle(p1: String, p2: String, p3: String, p4: String): Column = wrapExpression[ST_Angle](p1, p2, p3, p4)
+
+  def ST_Angle(p1: Column, p2: Column, p3: Column): Column = wrapExpression[ST_Angle](p1, p2, p3)
+
+  def ST_Angle(p1: String, p2: String, p3: String): Column = wrapExpression[ST_Angle](p1, p2, p3)
+
+  def ST_Angle(line1: Column, line2: Column): Column = wrapExpression[ST_Angle](line1, line2)
+
+  def ST_Angle(line1: String, line2: String): Column = wrapExpression[ST_Angle](line1, line2)
+
+  def ST_Degrees(angleInRadian: Column): Column = wrapExpression[ST_Degrees](angleInRadian)
+
+  def ST_Degrees(angleInRadian: Double): Column = wrapExpression[ST_Degrees](angleInRadian)
   def ST_HausdorffDistance(g1: Column, g2: Column) = wrapExpression[ST_HausdorffDistance](g1, g2, -1)
 
   def ST_HausdorffDistance(g1: String, g2: String) = wrapExpression[ST_HausdorffDistance](g1, g2, -1);
diff --git a/sql/common/src/test/scala/org/apache/sedona/sql/dataFrameAPITestScala.scala b/sql/common/src/test/scala/org/apache/sedona/sql/dataFrameAPITestScala.scala
index db85059865..f534bd7fc4 100644
--- a/sql/common/src/test/scala/org/apache/sedona/sql/dataFrameAPITestScala.scala
+++ b/sql/common/src/test/scala/org/apache/sedona/sql/dataFrameAPITestScala.scala
@@ -301,7 +301,7 @@ class dataFrameAPITestScala extends TestBaseScala {
       assert(actualResult == expectedResult)
     }
 
-    it("Passed ST_MakePolygon") {
+    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"))
       val actualResult = df.take(1)(0).get(0).asInstanceOf[Geometry].toText()
@@ -1035,6 +1035,38 @@ class dataFrameAPITestScala extends TestBaseScala {
       assertEquals(expected, actual)
     }
 
+    it("Passed ST_Angle - 4 Points") {
+      val polyDf = sparkSession.sql("SELECT ST_GeomFromWKT('POINT (10 10)') AS p1, ST_GeomFromWKT('POINT (0 0)') AS p2," +
+        " ST_GeomFromWKT('POINT (90 90)') AS p3, ST_GeomFromWKT('POINT (100 80)') AS p4")
+      val df = polyDf.select(ST_Angle("p1", "p2", "p3", "p4"))
+      val actualRad = df.take(1)(0).get(0).asInstanceOf[Double]
+      val dfDegrees = sparkSession.sql(s"SELECT ST_Degrees($actualRad)")
+      val actualDegrees = dfDegrees.take(1)(0).get(0).asInstanceOf[Double]
+      val expectedDegrees = 269.9999999999999
+      assertEquals(expectedDegrees, actualDegrees, 1e-9)
+    }
+
+    it("Passed ST_Angle - 3 Points") {
+      val polyDf = sparkSession.sql("SELECT ST_GeomFromWKT('POINT (0 0)') AS p1, ST_GeomFromWKT('POINT (10 10)') AS p2," +
+        " ST_GeomFromWKT('POINT (20 0)') AS p3")
+      val df = polyDf.select(ST_Angle("p1", "p2", "p3"))
+      val actualRad = df.take(1)(0).get(0).asInstanceOf[Double]
+      val dfDegrees = sparkSession.sql(s"SELECT ST_Degrees($actualRad)")
+      val actualDegrees = dfDegrees.take(1)(0).get(0).asInstanceOf[Double]
+      val expectedDegrees = 270
+      assertEquals(expectedDegrees, actualDegrees, 1e-9)
+    }
+
+    it("Passed ST_Angle - 2 LineStrings") {
+      val polyDf = sparkSession.sql("SELECT ST_GeomFromWKT('LINESTRING(0 0, 0.3 0.7, 1 1)') AS line1, ST_GeomFromWKT('LINESTRING(0 0, 0.2 0.5, 1 0)') AS line2")
+      val df = polyDf.select(ST_Angle("line1", "line2"))
+      val actualRad = df.take(1)(0).get(0).asInstanceOf[Double]
+      val dfDegrees = sparkSession.sql(s"SELECT ST_Degrees($actualRad)")
+      val actualDegrees = dfDegrees.take(1)(0).get(0).asInstanceOf[Double]
+      val expectedDegrees = 45
+      assertEquals(expectedDegrees, actualDegrees, 1e-9)
+    }
+
     it("Passed ST_HausdorffDistance") {
       val polyDf = sparkSession.sql("SELECT ST_GeomFromWKT('POLYGON ((1 2, 2 1, 2 0, 4 1, 1 2))') AS g1, " +
         "ST_GeomFromWKT('MULTILINESTRING ((1 1, 2 1, 4 4, 5 5), (10 10, 11 11, 12 12, 14 14), (-11 -20, -11 -21, -15 -19))') AS g2")
diff --git a/sql/common/src/test/scala/org/apache/sedona/sql/functionTestScala.scala b/sql/common/src/test/scala/org/apache/sedona/sql/functionTestScala.scala
index 5754de3de3..f413dc078e 100644
--- a/sql/common/src/test/scala/org/apache/sedona/sql/functionTestScala.scala
+++ b/sql/common/src/test/scala/org/apache/sedona/sql/functionTestScala.scala
@@ -23,6 +23,7 @@ import org.apache.commons.codec.binary.Hex
 import org.apache.sedona.sql.implicits._
 import org.apache.spark.sql.catalyst.expressions.{GenericRow, GenericRowWithSchema}
 import org.apache.spark.sql.functions._
+import org.apache.spark.sql.sedona_sql.expressions.ST_Degrees
 import org.apache.spark.sql.{DataFrame, Row}
 import org.geotools.referencing.CRS
 import org.junit.Assert.assertEquals
@@ -2017,7 +2018,6 @@ class functionTestScala extends TestBaseScala with Matchers with GeometrySample
       val actual = df.take(1)(0).get(0).asInstanceOf[Double]
       val expected = expectedResult
       assertEquals(expected, actual, 1e-9)
-
     }
   }
 
@@ -2045,7 +2045,7 @@ class functionTestScala extends TestBaseScala with Matchers with GeometrySample
 
   it ("should pass ST_BoundingDiagonal") {
     val geomTestCases = Map (
-      ("'POINT (10 10)'")-> "'LINESTRING (10 10, 10 10)'",
+      ("'POINT (10 10)'") -> "'LINESTRING (10 10, 10 10)'",
       ("'POLYGON ((1 1 1, 4 4 4, 0 9 3, 0 9 9, 1 1 1))'") -> "'LINESTRING Z(0 1 1, 4 9 9)'",
       ("'GEOMETRYCOLLECTION (MULTIPOLYGON (((1 1, 1 -1, 2 2, 2 9, 9 1, 1 1)), ((5 5, 4 4, 2 2 , 5 5))), POINT (-1 0))'") -> "'LINESTRING (-1 -1, 9 9)'"
     )
@@ -2057,6 +2057,67 @@ class functionTestScala extends TestBaseScala with Matchers with GeometrySample
     }
   }
 
+  it ("should pass ST_Angle - 4 points") {
+    val geomTestCases = Map (
+      ("'POINT (0 0)'", "'POINT (1 1)'", "'POINT (1 0)'", "'POINT (6 2)'") -> (0.4048917862850834, 23.198590513648185)
+    )
+    for (((geom), expectedResult) <- geomTestCases) {
+      val p1 = geom._1
+      val p2 = geom._2
+      val p3 = geom._3
+      val p4 = geom._4
+      val df = sparkSession.sql(s"SELECT ST_Angle(ST_GeomFromWKT($p1), ST_GeomFromWKT($p2), ST_GeomFromWKT($p3), ST_GeomFromWKT($p4)) AS angleInRadian")
+      val expectedRadian = expectedResult._1
+      val expectedDegrees = expectedResult._2
+
+      val actualRadian = df.take(1)(0).get(0).asInstanceOf[Double]
+      val actualDegrees = df.selectExpr("ST_Degrees(angleInRadian)").take(1)(0).get(0).asInstanceOf[Double]
+
+      assertEquals(expectedRadian, actualRadian, 1e-9)
+      assertEquals(expectedDegrees, actualDegrees, 1e-9)
+    }
+  }
+
+  it ("should pass ST_Angle - 3 points") {
+    val geomTestCases = Map(
+      ("'POINT (1 1)'", "'POINT (0 0)'", "'POINT (3 2)'") -> (0.19739555984988044, 11.309932474020195)
+    )
+    for (((geom), expectedResult) <- geomTestCases) {
+      val p1 = geom._1
+      val p2 = geom._2
+      val p3 = geom._3
+      val df = sparkSession.sql(s"SELECT ST_Angle(ST_GeomFromWKT($p1), ST_GeomFromWKT($p2), ST_GeomFromWKT($p3)) AS angleInRadian")
+      val expectedRadian = expectedResult._1
+      val expectedDegrees = expectedResult._2
+
+      val actualRadian = df.take(1)(0).get(0).asInstanceOf[Double]
+      val actualDegrees = df.selectExpr("ST_Degrees(angleInRadian)").take(1)(0).get(0).asInstanceOf[Double]
+
+      assertEquals(expectedRadian, actualRadian, 1e-9)
+      assertEquals(expectedDegrees, actualDegrees, 1e-9)
+    }
+  }
+
+  it ("should pass ST_Angle - 2 lines") {
+    val geomTestCases = Map(
+      ("'LINESTRING (0 0, 1 1)'", "'LINESTRING (0 0, 3 2)'") -> (0.19739555984988044, 11.309932474020195)
+    )
+    for (((geom), expectedResult) <- geomTestCases) {
+      val p1 = geom._1
+      val p2 = geom._2
+
+      val df = sparkSession.sql(s"SELECT ST_Angle(ST_GeomFromWKT($p1), ST_GeomFromWKT($p2)) AS angleInRadian")
+      val expectedRadian = expectedResult._1
+      val expectedDegrees = expectedResult._2
+
+      val actualRadian = df.take(1)(0).get(0).asInstanceOf[Double]
+      val actualDegrees = df.selectExpr("ST_Degrees(angleInRadian)").take(1)(0).get(0).asInstanceOf[Double]
+
+      assertEquals(expectedRadian, actualRadian, 1e-9)
+      assertEquals(expectedDegrees, actualDegrees, 1e-9)
+    }
+  }
+
   it ("should pass ST_HausdorffDistance") {
     val geomTestCases = Map (
       ("'LINESTRING (1 2, 1 5, 2 6, 1 2)'", "'POINT (10 34)'", 0.34) -> (33.24154027718932, 33.24154027718932),
@@ -2077,7 +2138,7 @@ class functionTestScala extends TestBaseScala with Matchers with GeometrySample
       assert(expectedDefaultValue == actualDefaultValue)
     }
   }
-  
+
   it ("should pass GeometryType") {
     val geomTestCases = Map (
       ("'POINT (51.3168 -0.56)'") -> "'POINT'",
@@ -2096,5 +2157,5 @@ class functionTestScala extends TestBaseScala with Matchers with GeometrySample
         assertEquals(expected, actual)
     }
   }
-
+  
 }