[SPARK-54142][GEO][SQL][PYTHON] Implement the st_srid function in Scala and PySpark

### What changes were proposed in this pull request?
Implement the `st_srid` function in Scala and PySpark API.

### Why are the changes needed?
Expand API support for the `ST_Srid` expression.

### Does this PR introduce _any_ user-facing change?
Yes, the new function is now available in Scala and PySpark API.

### How was this patch tested?
Added appropriate Scala function unit tests:
- `STFunctionsSuite`

Added appropriate PySpark function unit tests:
- `test_functions`

### Was this patch authored or co-authored using generative AI tooling?
No.

Closes #52841 from uros-db/geo-ST_Srid-scala.

Authored-by: Uros Bojanic <uros.bojanic@databricks.com>
Signed-off-by: Hyukjin Kwon <gurwls223@apache.org>

Author: uros-db

python/docs/source/reference/pyspark.sql/functions.rst

@@ -660,6 +660,7 @@ Geospatial ST Functions
     st_asbinary
     st_geogfromwkb
     st_geomfromwkb
+    st_srid
 
 
 UDF, UDTF and UDT

python/pyspark/sql/connect/functions/builtin.py

@@ -4807,6 +4807,13 @@ def st_geomfromwkb(wkb: "ColumnOrName") -> Column:
 st_geomfromwkb.__doc__ = pysparkfuncs.st_geomfromwkb.__doc__
 
 
+def st_srid(geo: "ColumnOrName") -> Column:
+    return _invoke_function_over_columns("st_srid", geo)
+
+
+st_srid.__doc__ = pysparkfuncs.st_srid.__doc__
+
+
 # Call Functions
 
 

python/pyspark/sql/functions/__init__.py

@@ -526,6 +526,7 @@
     "st_asbinary",
     "st_geogfromwkb",
     "st_geomfromwkb",
+    "st_srid",
     # Call Functions
     "call_udf",
     "pandas_udf",

python/pyspark/sql/functions/builtin.py

@@ -25971,6 +25971,31 @@ def st_geomfromwkb(wkb: "ColumnOrName") -> Column:
     return _invoke_function_over_columns("st_geomfromwkb", wkb)
 
 
+@_try_remote_functions
+def st_srid(geo: "ColumnOrName") -> Column:
+    """Returns the SRID of the input GEOGRAPHY or GEOMETRY value.
+
+    .. versionadded:: 4.1.0
+
+    Parameters
+    ----------
+    geo : :class:`~pyspark.sql.Column` or str
+        A geospatial value, either a GEOGRAPHY or a GEOMETRY.
+
+    Examples
+    --------
+    >>> from pyspark.sql import functions as sf
+    >>> df = spark.createDataFrame([(bytes.fromhex('0101000000000000000000F03F0000000000000040'),)], ['wkb'])  # noqa
+    >>> df.select(sf.st_srid(sf.st_geogfromwkb('wkb')).alias('result')).collect()
+    [Row(result=4326)]
+    >>> from pyspark.sql import functions as sf
+    >>> df = spark.createDataFrame([(bytes.fromhex('0101000000000000000000F03F0000000000000040'),)], ['wkb'])  # noqa
+    >>> df.select(sf.st_srid(sf.st_geomfromwkb('wkb')).alias('result')).collect()
+    [Row(result=0)]
+    """
+    return _invoke_function_over_columns("st_srid", geo)
+
+
 # Call Functions
 
 

python/pyspark/sql/tests/test_functions.py

@@ -2822,6 +2822,21 @@ def test_st_asbinary(self):
         )
         self.assertEqual(results, [expected])
 
+    def test_st_srid(self):
+        df = self.spark.createDataFrame(
+            [(bytes.fromhex("0101000000000000000000F03F0000000000000040"),)],
+            ["wkb"],
+        )
+        results = df.select(
+            F.st_srid(F.st_geogfromwkb("wkb")),
+            F.st_srid(F.st_geomfromwkb("wkb")),
+        ).collect()
+        expected = Row(
+            4326,
+            0,
+        )
+        self.assertEqual(results, [expected])
+
 
 class FunctionsTests(ReusedSQLTestCase, FunctionsTestsMixin):
     pass

sql/api/src/main/scala/org/apache/spark/sql/functions.scala

@@ -9171,6 +9171,15 @@ object functions {
   def st_geomfromwkb(wkb: Column): Column =
     Column.fn("st_geomfromwkb", wkb)
 
+  /**
+   * Returns the SRID of the input GEOGRAPHY or GEOMETRY value.
+   *
+   * @group st_funcs
+   * @since 4.1.0
+   */
+  def st_srid(geo: Column): Column =
+    Column.fn("st_srid", geo)
+
   //////////////////////////////////////////////////////////////////////////////////////////////
   // Scala UDF functions
   //////////////////////////////////////////////////////////////////////////////////////////////

sql/core/src/test/scala/org/apache/spark/sql/STFunctionsSuite.scala

@@ -42,4 +42,20 @@ class STFunctionsSuite extends QueryTest with SharedSparkSession {
         "0101000000000000000000f03f0000000000000040"))
   }
 
+  /** ST accessor expressions. */
+
+  test("st_srid") {
+    // Test data: Well-Known Binary (WKB) representations.
+    val df = Seq[(String)](
+      (
+        "0101000000000000000000f03f0000000000000040"
+      )).toDF("wkb")
+    // ST_GeogFromWKB/ST_GeomFromWKB and ST_Srid.
+    checkAnswer(
+      df.select(
+        st_srid(st_geogfromwkb(unhex($"wkb"))).as("col0"),
+        st_srid(st_geomfromwkb(unhex($"wkb"))).as("col1")),
+      Row(4326, 0))
+  }
+
 }