[SPARK-57397][PYTHON][3.5] Raise an exception when parsed UDT is the wrong type

python/pyspark/errors/error_classes.py

@@ -231,6 +231,11 @@
       "Duplicated field names in Arrow Struct are not allowed, got <field_names>"
     ]
   },
+  "FIELD_TYPE_MISMATCH": {
+    "message": [
+      "<obj> is not an instance of type <data_type>."
+    ]
+  },
   "HIGHER_ORDER_FUNCTION_SHOULD_RETURN_COLUMN" : {
     "message" : [
       "Function `<func_name>` should return Column, got <return_type>."

python/pyspark/sql/tests/test_types.py

@@ -49,6 +49,7 @@
     BinaryType,
     BooleanType,
     NullType,
+    UserDefinedType,
 )
 from pyspark.sql.types import (
     _array_signed_int_typecode_ctype_mappings,
@@ -541,6 +542,16 @@ def check_datatype(datatype):
         _make_type_verifier(PythonOnlyUDT())(PythonOnlyPoint(1.0, 2.0))
         self.assertRaises(ValueError, lambda: _make_type_verifier(PythonOnlyUDT())([1.0, 2.0]))
 
+    def test_udt_from_json_import_type_mismatch(self):
+        json = {
+            "type": "udt",
+            "pyClass": "random.random",
+            "serializedClass": "",
+            "sqlType": StringType().jsonValue(),
+        }
+        with self.assertRaises(PySparkTypeError):
+            UserDefinedType.fromJson(json)
+
     def test_simple_udt_in_df(self):
         schema = StructType().add("key", LongType()).add("val", PythonOnlyUDT())
         df = self.spark.createDataFrame(

python/pyspark/sql/types.py

@@ -1210,6 +1210,11 @@ def fromJson(cls, json: Dict[str, Any]) -> "UserDefinedType":
             )
         else:
             UDT = getattr(m, pyClass)
+            if not (isinstance(UDT, type) and issubclass(UDT, UserDefinedType)):
+                raise PySparkTypeError(
+                    error_class="FIELD_TYPE_MISMATCH",
+                    message_parameters={"obj": str(UDT), "data_type": "UserDefinedType"},
+                )
         return UDT()
 
     def __eq__(self, other: Any) -> bool: