[SPARK-53957][GEO][SQL] Support GEOGRAPHY and GEOMETRY in the SpatialReferenceSystemMapper

### What changes were proposed in this pull request?
Extend the `SpatialReferenceSystemMapper` class to offer SRID <-> CRS mappings for both `GeographyType` and `GeometryType`.

The `SpatialReferenceSystemMapper` class was introduced originally as part of: apache#52491.

### Why are the changes needed?
Avoid manual checks for GEOGRAPHY type, and use the centralized SRS mapping logic.

Also, this PR will make it easier to support additional SRID/CRS values for spatial types soon.

### Does this PR introduce _any_ user-facing change?
No.

### How was this patch tested?
Added new tests to:
- `SpatialReferenceSystemMapperSuite`

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

Closes apache#52667 from uros-db/geo-srsMapper.

Authored-by: Uros Bojanic <[email protected]>
Signed-off-by: Wenchen Fan <[email protected]>

Author: uros-db

sql/api/src/main/scala/org/apache/spark/sql/internal/types/CartesianSpatialReferenceSystemMapper.java

@@ -0,0 +1,35 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.internal.types;
+
+/**
+ * Class for providing SRS mappings for cartesian spatial reference systems.
+ */
+public class CartesianSpatialReferenceSystemMapper extends SpatialReferenceSystemMapper {
+  // Returns the string ID corresponding to the input SRID. If not supported, returns `null`.
+  public static String getStringId(int srid) {
+    SpatialReferenceSystemInformation srsInfo = srsCache.getSrsInfo(srid);
+    return srsInfo != null ? srsInfo.stringId() : null;
+  }
+
+  // Returns the SRID corresponding to the input string ID. If not supported, returns `null`.
+  public static Integer getSrid(String stringId) {
+    SpatialReferenceSystemInformation srsInfo = srsCache.getSrsInfo(stringId);
+    return srsInfo != null ? srsInfo.srid() : null;
+  }
+}

sql/api/src/main/scala/org/apache/spark/sql/internal/types/GeographicSpatialReferenceSystemMapper.java

@@ -0,0 +1,35 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.internal.types;
+
+/**
+ * Class for providing SRS mappings for geographic spatial reference systems.
+ */
+public class GeographicSpatialReferenceSystemMapper extends SpatialReferenceSystemMapper {
+  // Returns the string ID corresponding to the input SRID. If not supported, returns `null`.
+  public static String getStringId(int srid) {
+    SpatialReferenceSystemInformation srsInfo = srsCache.getSrsInfo(srid);
+    return srsInfo != null && srsInfo.isGeographic() ? srsInfo.stringId() : null;
+  }
+
+  // Returns the SRID corresponding to the input string ID. If not supported, returns `null`.
+  public static Integer getSrid(String stringId) {
+    SpatialReferenceSystemInformation srsInfo = srsCache.getSrsInfo(stringId);
+    return srsInfo != null && srsInfo.isGeographic() ? srsInfo.srid() : null;
+  }
+}

sql/api/src/main/scala/org/apache/spark/sql/internal/types/SpatialReferenceSystemCache.java

@@ -0,0 +1,84 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.internal.types;
+
+import java.util.HashMap;
+import java.util.List;
+
+/**
+ * Class for maintaining the mappings between supported SRID/CRS values and the corresponding SRS.
+ */
+public class SpatialReferenceSystemCache {
+
+  // Private constructor to prevent external instantiation of this singleton class.
+  private SpatialReferenceSystemCache() {
+    populateSpatialReferenceSystemInformationMapping();
+  }
+
+  // The singleton `instance` is created lazily, meaning that it is not instantiated until the
+  // `getInstance()` method is called for the first time. Note that this solution is thread-safe.
+  private static volatile SpatialReferenceSystemCache instance = null;
+
+  // The `getInstance` method uses double-checked locking to ensure efficient and safe instance
+  // creation. The singleton instance is created only once, even in a multithreaded environment.
+  public static SpatialReferenceSystemCache getInstance() {
+    if (instance == null) {
+      synchronized (SpatialReferenceSystemCache.class) {
+        if (instance == null) {
+          instance = new SpatialReferenceSystemCache();
+        }
+      }
+    }
+    return instance;
+  }
+
+  // Hash map for defining the mappings from the integer SRID value to the full SRS information.
+  private final HashMap<Integer, SpatialReferenceSystemInformation> sridToSrs =
+      new HashMap<>();
+
+  // Hash map for defining the mappings from the string ID value to the full SRS information.
+  private final HashMap<String, SpatialReferenceSystemInformation> stringIdToSrs =
+      new HashMap<>();
+
+  // Helper method for building the SRID-to-SRS and stringID-to-SRS mappings.
+  private void populateSpatialReferenceSystemInformationMapping() {
+    // Currently, we only support a limited set of SRID / CRS values. However, we will soon extend
+    // this to support all the SRIDs supported by relevant authorities and libraries. The SRS list
+    // below will be updated accordingly, and the maps will be populated with more complete data.
+    List<SpatialReferenceSystemInformation> srsInformationList = List.of(
+      new SpatialReferenceSystemInformation(0, "SRID:0", false),
+      new SpatialReferenceSystemInformation(3857, "EPSG:3857", false),
+      new SpatialReferenceSystemInformation(4326, "OGC:CRS84", true)
+    );
+    // Populate the mappings using the same SRS information objects, avoiding any duplication.
+    for (SpatialReferenceSystemInformation srsInformation: srsInformationList) {
+      sridToSrs.put(srsInformation.srid(), srsInformation);
+      stringIdToSrs.put(srsInformation.stringId(), srsInformation);
+    }
+  }
+
+  // Returns the SRS corresponding to the input SRID. If not supported, returns `null`.
+  public SpatialReferenceSystemInformation getSrsInfo(int srid) {
+    return sridToSrs.getOrDefault(srid, null);
+  }
+
+  // Returns the SRS corresponding to the input string ID. If not supported, returns `null`.
+  public SpatialReferenceSystemInformation getSrsInfo(String stringId) {
+    return stringIdToSrs.getOrDefault(stringId, null);
+  }
+}

sql/api/src/main/scala/org/apache/spark/sql/internal/types/SpatialReferenceSystemInformation.java

@@ -0,0 +1,30 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.internal.types;
+
+/**
+ * Class for maintaining information about a spatial reference system (SRS).
+ */
+public record SpatialReferenceSystemInformation(
+  // Field storing the spatial reference identifier (SRID) value of this SRS.
+  int srid,
+  // Field storing the string ID of the corresponding coordinate reference system (CRS).
+  String stringId,
+  // Field indicating whether the spatial reference system (SRS) is geographic or not.
+  boolean isGeographic
+) {}

sql/api/src/main/scala/org/apache/spark/sql/internal/types/SpatialReferenceSystemMapper.java

@@ -17,59 +17,10 @@
 
 package org.apache.spark.sql.internal.types;
 
-import java.util.HashMap;
-
-/*
- * Class for maintaining mappings between supported SRID values and the string ID of the
- * corresponding CRS.
+/**
+ * Abstract class for providing SRS mappings for spatial reference systems.
  */
-public class SpatialReferenceSystemMapper {
-
-  // We implement this class as a singleton (we disallow construction).
-  private SpatialReferenceSystemMapper() {}
-
-  private static final SpatialReferenceSystemMapper Instance = new SpatialReferenceSystemMapper();
-
-  // Returns the unique instance of this class.
-  public static SpatialReferenceSystemMapper get() {
-    return Instance;
-  }
-
-  // Hash maps defining the mappings to/from SRID and string ID for a CRS.
-  private static final HashMap<Integer, String> sridToStringId = buildSridToStringIdMap();
-  private static final HashMap<String, Integer> stringIdToSrid = buildStringIdToSridMap();
-
-  // Returns the string ID corresponding to the input SRID. If the input SRID is not supported,
-  // `null` is returned.
-  public String getStringId(int srid) {
-    return sridToStringId.get(srid);
-  }
-
-  // Returns the SRID corresponding to the input string ID. If the input string ID is not
-  // supported, `null` is returned.
-  public Integer getSrid(String stringId) {
-    return stringIdToSrid.get(stringId);
-  }
-
-  // Currently, we only support a limited set of SRID / CRS mappings. However, we will soon extend
-  // this to support all the SRIDs supported by relevant authorities and libraries. The methods
-  // below will be updated accordingly, in order to populate the mappings with more complete data.
-
-  // Helper method for building the SRID-to-string-ID mapping.
-  private static HashMap<Integer, String> buildSridToStringIdMap() {
-    HashMap<Integer, String> map = new HashMap<>();
-    map.put(0, "SRID:0"); // Unspecified
-    map.put(3857, "EPSG:3857"); // Web Mercator
-    map.put(4326, "OGC:CRS84"); // WGS84
-    return map;
-  }
-
-  // Helper method for building the string-ID-to-SRID mapping.
-  private static HashMap<String, Integer> buildStringIdToSridMap() {
-    HashMap<String, Integer> map = new HashMap<>();
-    map.put("SRID:0", 0); // Unspecified
-    map.put("EPSG:3857", 3857); // Web Mercator
-    map.put("OGC:CRS84", 4326); // WGS84
-    return map;
-  }
+public abstract class SpatialReferenceSystemMapper {
+  protected static final SpatialReferenceSystemCache srsCache =
+      SpatialReferenceSystemCache.getInstance();
 }

sql/api/src/main/scala/org/apache/spark/sql/types/GeographyType.scala

@@ -21,6 +21,7 @@ import org.json4s.JsonAST.{JString, JValue}
 
 import org.apache.spark.SparkIllegalArgumentException
 import org.apache.spark.annotation.Experimental
+import org.apache.spark.sql.internal.types.GeographicSpatialReferenceSystemMapper
 
 /**
  * The data type representing GEOGRAPHY values which are spatial objects, as defined in the Open
@@ -160,8 +161,8 @@ object GeographyType extends SpatialType {
    * Constructors for GeographyType.
    */
   def apply(srid: Int): GeographyType = {
-    if (!isValidSrid(srid)) {
-      // Limited geographic SRID values are allowed.
+    val crs = GeographicSpatialReferenceSystemMapper.getStringId(srid)
+    if (crs == null) {
       throw new SparkIllegalArgumentException(
         errorClass = "ST_INVALID_SRID_VALUE",
         messageParameters = Map("srid" -> srid.toString))
@@ -193,35 +194,9 @@ object GeographyType extends SpatialType {
   }
 
   def apply(crs: String, algorithm: EdgeInterpolationAlgorithm): GeographyType = {
-    if (!isValidCrs(crs)) {
-      // Limited geographic CRS values are allowed.
-      throw new SparkIllegalArgumentException(
-        errorClass = "ST_INVALID_CRS_VALUE",
-        messageParameters = Map("crs" -> crs))
-    }
     new GeographyType(crs, algorithm)
   }
 
-  /**
-   * Helper method to validate the CRS value. Limited geographic CRS values are allowed.
-   */
-  private def isValidCrs(crs: String): Boolean = {
-    // Currently, we only support "OGC:CRS84" / "EPSG:4326" / "SRID:ANY".
-    // In the future, we may support others.
-    crs.equalsIgnoreCase(GEOGRAPHY_DEFAULT_CRS) ||
-    crs.equalsIgnoreCase(GEOGRAPHY_DEFAULT_EPSG_CRS) ||
-    crs.equalsIgnoreCase(MIXED_CRS)
-  }
-
-  /**
-   * Helper method to validate the SRID value. Only geographic SRID values are allowed.
-   */
-
-  private def isValidSrid(srid: Int): Boolean = {
-    // Currently, we only support 4326. In the future, we may support others.
-    srid == GEOGRAPHY_DEFAULT_SRID
-  }
-
   override private[sql] def defaultConcreteType: DataType = GEOGRAPHY_MIXED_TYPE
 
   override private[sql] def acceptsType(other: DataType): Boolean =
@@ -235,17 +210,17 @@ object GeographyType extends SpatialType {
   private[types] def toSrid(crs: String): Int = {
     // The special value "SRID:ANY" is used to represent mixed SRID values.
     if (crs.equalsIgnoreCase(GeographyType.MIXED_CRS)) {
-      GeographyType.MIXED_SRID
+      return GeographyType.MIXED_SRID
     }
-    // As for other valid CRS values, we currently offer limited support.
-    else if (crs.equalsIgnoreCase(GeographyType.GEOGRAPHY_DEFAULT_CRS) ||
-      crs.equalsIgnoreCase(GeographyType.GEOGRAPHY_DEFAULT_EPSG_CRS)) {
-      GeographyType.GEOGRAPHY_DEFAULT_SRID
-    } else {
+    // For all other CRS values, we need to look up the corresponding SRID.
+    val srid = GeographicSpatialReferenceSystemMapper.getSrid(crs)
+    if (srid == null) {
+      // If the CRS value is not recognized, we throw an exception.
       throw new SparkIllegalArgumentException(
         errorClass = "ST_INVALID_CRS_VALUE",
         messageParameters = Map("crs" -> crs))
     }
+    srid
   }
 }
 

sql/api/src/main/scala/org/apache/spark/sql/types/GeometryType.scala

@@ -21,7 +21,7 @@ import org.json4s.JsonAST.{JString, JValue}
 
 import org.apache.spark.SparkIllegalArgumentException
 import org.apache.spark.annotation.Experimental
-import org.apache.spark.sql.internal.types.SpatialReferenceSystemMapper
+import org.apache.spark.sql.internal.types.CartesianSpatialReferenceSystemMapper
 
 /**
  * The data type representing GEOMETRY values which are spatial objects, as defined in the Open
@@ -153,7 +153,7 @@ object GeometryType extends SpatialType {
    * Constructors for GeometryType.
    */
   def apply(srid: Int): GeometryType = {
-    val crs = SpatialReferenceSystemMapper.get().getStringId(srid)
+    val crs = CartesianSpatialReferenceSystemMapper.getStringId(srid)
     if (crs == null) {
       throw new SparkIllegalArgumentException(
         errorClass = "ST_INVALID_SRID_VALUE",
@@ -191,7 +191,7 @@ object GeometryType extends SpatialType {
       return GeometryType.MIXED_SRID
     }
     // For all other CRS values, we need to look up the corresponding SRID.
-    val srid = SpatialReferenceSystemMapper.get().getSrid(crs)
+    val srid = CartesianSpatialReferenceSystemMapper.getSrid(crs)
     if (srid == null) {
       // If the CRS value is not recognized, we throw an exception.
       throw new SparkIllegalArgumentException(