diff --git a/python/pyspark/sql/tests/test_geographytype.py b/python/pyspark/sql/tests/test_geographytype.py
index 701ee78cd7ad..43dd7b8c321d 100644
--- a/python/pyspark/sql/tests/test_geographytype.py
+++ b/python/pyspark/sql/tests/test_geographytype.py
@@ -78,6 +78,47 @@ def test_geographytype_different_srid_values(self):
             geography_type_2 = GeographyType("ANY")
             self.assertNotEqual(geography_type_1.srid, geography_type_2.srid)
 
+    def test_geographytype_from_invalid_crs(self):
+        """Test that GeographyType construction fails when an invalid CRS is specified."""
+
+        for invalid_crs in ["srid", "any", "ogccrs84", "ogc:crs84", "ogc:CRS84", "asdf", ""]:
+            with self.assertRaises(IllegalArgumentException) as error_context:
+                GeographyType._from_crs(invalid_crs, "SPHERICAL")
+            crs_header = "[ST_INVALID_CRS_VALUE] Invalid or unsupported CRS"
+            self.assertEqual(
+                str(error_context.exception),
+                f"{crs_header} (coordinate reference system) value: '{invalid_crs}'.",
+            )
+
+    # The tests below verify GEOGRAPHY type JSON parsing based on CRS and algorithm.
+
+    def test_geographytype_from_invalid_algorithm(self):
+        """Test that GeographyType construction fails when an invalid CRS is specified."""
+
+        for invalid_alg in ["alg", "algorithm", "KARNEY", "spherical", "SPHEROID", "asdf", ""]:
+            with self.assertRaises(IllegalArgumentException) as error_context:
+                GeographyType._from_crs("OGC:CRS84", invalid_alg)
+            alg_header = "[ST_INVALID_ALGORITHM_VALUE] Invalid or unsupported"
+            self.assertEqual(
+                str(error_context.exception),
+                f"{alg_header} edge interpolation algorithm value: '{invalid_alg}'.",
+            )
+
+    def test_geographytype_from_valid_crs_and_algorithm(self):
+        """Test that GeographyType construction passes when valid CRS & ALG are specified."""
+
+        supported_crs = {
+            "OGC:CRS84": 4326,
+        }
+        for valid_crs, srid in supported_crs.items():
+            for valid_alg in ["SPHERICAL"]:
+                geography_type = GeographyType._from_crs(valid_crs, valid_alg)
+                self.assertEqual(geography_type.srid, srid)
+                self.assertEqual(geography_type.typeName(), "geography")
+                self.assertEqual(geography_type.simpleString(), f"geography({srid})")
+                self.assertEqual(geography_type.jsonValue(), f"geography({valid_crs}, {valid_alg})")
+                self.assertEqual(repr(geography_type), f"GeographyType({srid})")
+
 
 class GeographyTypeTest(GeographyTypeTestMixin, ReusedSQLTestCase):
     pass
diff --git a/python/pyspark/sql/tests/test_geometrytype.py b/python/pyspark/sql/tests/test_geometrytype.py
index 8647404d4f88..764b5d6b45de 100644
--- a/python/pyspark/sql/tests/test_geometrytype.py
+++ b/python/pyspark/sql/tests/test_geometrytype.py
@@ -78,6 +78,34 @@ def test_geometrytype_different_srid_values(self):
             geometry_type_2 = GeometryType("ANY")
             self.assertNotEqual(geometry_type_1.srid, geometry_type_2.srid)
 
+    # The tests below verify GEOMETRY type JSON parsing based on the CRS value.
+
+    def test_geometrytype_from_invalid_crs(self):
+        """Test that GeometryType construction fails when an invalid CRS is specified."""
+
+        for invalid_crs in ["srid", "any", "ogccrs84", "ogc:crs84", "ogc:CRS84", "asdf", ""]:
+            with self.assertRaises(IllegalArgumentException) as error_context:
+                GeometryType._from_crs(invalid_crs)
+            crs_header = "[ST_INVALID_CRS_VALUE] Invalid or unsupported CRS"
+            self.assertEqual(
+                str(error_context.exception),
+                f"{crs_header} (coordinate reference system) value: '{invalid_crs}'.",
+            )
+
+    def test_geometrytype_from_valid_crs(self):
+        """Test that GeometryType construction passes when a valid CRS is specified."""
+
+        supported_crs = {
+            "OGC:CRS84": 4326,
+        }
+        for valid_crs, srid in supported_crs.items():
+            geometry_type = GeometryType._from_crs(valid_crs)
+            self.assertEqual(geometry_type.srid, srid)
+            self.assertEqual(geometry_type.typeName(), "geometry")
+            self.assertEqual(geometry_type.simpleString(), f"geometry({srid})")
+            self.assertEqual(geometry_type.jsonValue(), f"geometry({valid_crs})")
+            self.assertEqual(repr(geometry_type), f"GeometryType({srid})")
+
 
 class GeometryTypeTest(GeometryTypeTestMixin, ReusedSQLTestCase):
     pass
diff --git a/python/pyspark/sql/types.py b/python/pyspark/sql/types.py
index 1295936be33b..f45822bfa6d2 100644
--- a/python/pyspark/sql/types.py
+++ b/python/pyspark/sql/types.py
@@ -567,7 +567,7 @@ def _from_crs(cls, crs: str, alg: str) -> "GeographyType":
         # Algorithm value must be validated, although only SPHERICAL is supported currently.
         if alg != cls.DEFAULT_ALG:
             raise IllegalArgumentException(
-                errorClass="INVALID_ALGORITHM_VALUE",
+                errorClass="ST_INVALID_ALGORITHM_VALUE",
                 messageParameters={
                     "alg": str(alg),
                 },