[SPARK-56584][PYTHON] Generalize RESULT_TYPE_MISMATCH_FOR_ARROW_UDF error class and remove dead SCHEMA_MISMATCH_FOR_ARROW_PYTHON_UDF

python/pyspark/errors/error-conditions.json

@@ -808,26 +808,21 @@
       "Number of columns of the returned data doesn't match specified schema. Expected: <expected> Actual: <actual>"
     ]
   },
-  "RESULT_ROWS_MISMATCH": {
+  "RESULT_COLUMN_TYPES_MISMATCH": {
     "message": [
-      "The number of output rows (<output_length>) must match the number of input rows (<input_length>)."
+      "Column types of the returned data do not match specified schema. Mismatch: <mismatch>."
     ]
   },
-  "RESULT_TYPE_MISMATCH_FOR_ARROW_UDF": {
+  "RESULT_ROWS_MISMATCH": {
     "message": [
-      "Columns do not match in their data type: <mismatch>."
+      "The number of output rows (<output_length>) must match the number of input rows (<input_length>)."
     ]
   },
   "REUSE_OBSERVATION": {
     "message": [
       "An Observation can be used with a DataFrame only once."
     ]
   },
-  "SCHEMA_MISMATCH_FOR_ARROW_PYTHON_UDF": {
-    "message": [
-      "Result vector from <udf_type> was not the required length: expected <expected>, got <actual>."
-    ]
-  },
   "SCHEMA_MISMATCH_FOR_PANDAS_UDF": {
     "message": [
       "Result vector from <udf_type> was not the required length: expected <expected>, got <actual>."

python/pyspark/sql/tests/arrow/test_arrow_cogrouped_map.py

@@ -147,7 +147,8 @@ def test_apply_in_arrow_returning_wrong_types(self):
                 with self.quiet():
                     with self.assertRaisesRegex(
                         PythonException,
-                        f"Columns do not match in their data type: {expected}",
+                        "Column types of the returned data do not match specified schema. "
+                        f"Mismatch: {expected}",
                     ):
                         self.cogrouped.applyInArrow(
                             lambda left, right: left, schema=schema
@@ -171,7 +172,8 @@ def test_apply_in_arrow_returning_wrong_types_positional_assignment(self):
                     with self.quiet():
                         with self.assertRaisesRegex(
                             PythonException,
-                            f"Columns do not match in their data type: {expected}",
+                            "Column types of the returned data do not match specified schema. "
+                            f"Mismatch: {expected}",
                         ):
                             self.cogrouped.applyInArrow(
                                 lambda left, right: left, schema=schema

python/pyspark/sql/tests/arrow/test_arrow_grouped_map.py

@@ -171,7 +171,8 @@ def test_apply_in_arrow_returning_wrong_types(self):
                     for func_variation in function_variations(lambda table: table):
                         with self.assertRaisesRegex(
                             PythonException,
-                            f"Columns do not match in their data type: {expected}",
+                            "Column types of the returned data do not match specified schema. "
+                            f"Mismatch: {expected}",
                         ):
                             df.groupby("id").applyInArrow(func_variation, schema=schema).collect()
 
@@ -196,7 +197,8 @@ def test_apply_in_arrow_returning_wrong_types_positional_assignment(self):
                         for func_variation in function_variations(lambda table: table):
                             with self.assertRaisesRegex(
                                 PythonException,
-                                f"Columns do not match in their data type: {expected}",
+                                "Column types of the returned data do not match specified schema. "
+                                f"Mismatch: {expected}",
                             ):
                                 df.groupby("id").applyInArrow(
                                     func_variation, schema=schema

python/pyspark/worker.py

@@ -612,7 +612,7 @@ def verify_arrow_result(result, assign_cols_by_name, expected_cols_and_types):
 
         if type_mismatch:
             raise PySparkRuntimeError(
-                errorClass="RESULT_TYPE_MISMATCH_FOR_ARROW_UDF",
+                errorClass="RESULT_COLUMN_TYPES_MISMATCH",
                 messageParameters={
                     "mismatch": ", ".join(
                         "column '{}' (expected {}, actual {})".format(name, expected, actual)