[SPARK-48961][PYTHON] Make the parameter naming of PySparkException consistent with JVM

### What changes were proposed in this pull request?

This PR proposes to make the parameter naming of `PySparkException` consistent with JVM

### Why are the changes needed?

The parameter names of `PySparkException` are different from `SparkException` so there is an inconsistency when searching those parameters from error logs.

SparkException:
https://github.com/apache/spark/blob/6508b1f5e18731359354af0a7bcc1382bc4f356b/common/utils/src/main/scala/org/apache/spark/SparkException.scala#L27-L33

PySparkException:
https://github.com/apache/spark/blob/6508b1f5e18731359354af0a7bcc1382bc4f356b/python/pyspark/errors/exceptions/base.py#L29-L40

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

The error parameter names are changed from:
- `error_class` -> `errorClass`
- `message_parameters` -> `messageParameters`
- `query_contexts` -> `context`

### How was this patch tested?

The existing CI should pass

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

No.

Closes apache#47436 from itholic/SPARK-48961.

Authored-by: Haejoon Lee <haejoon.lee@databricks.com>
Signed-off-by: Haejoon Lee <haejoon.lee@databricks.com>

Author: itholic

python/docs/source/development/contributing.rst

@@ -279,8 +279,8 @@ Throw with arbitrary error message:
 .. code-block:: python
 
   class PySparkTestError(PySparkException):
-      def __init__(self, error_class: str, message_parameters: Dict[str, str]):
-          super().__init__(error_class=error_class, message_parameters=message_parameters)
+      def __init__(self, errorClass: str, messageParameters: Dict[str, str]):
+          super().__init__(errorClass=errorClass, messageParameters=messageParameters)
   
       def getMessageParameters(self) -> Optional[Dict[str, str]]:
           return super().getMessageParameters()

python/docs/source/development/logger.rst

@@ -52,7 +52,7 @@ Example log entry:
         "file": "/path/to/file.py",
         "line": "17",
         "fragment": "divide"
-        "error_class": "DIVIDE_BY_ZERO"
+        "errorClass": "DIVIDE_BY_ZERO"
       },
       "exception": {
         "class": "Py4JJavaError",

python/pyspark/accumulators.py

@@ -149,8 +149,8 @@ def value(self) -> T:
         """Get the accumulator's value; only usable in driver program"""
         if self._deserialized:
             raise PySparkRuntimeError(
-                error_class="VALUE_NOT_ACCESSIBLE",
-                message_parameters={
+                errorClass="VALUE_NOT_ACCESSIBLE",
+                messageParameters={
                     "value": "Accumulator.value",
                 },
             )
@@ -161,8 +161,8 @@ def value(self, value: T) -> None:
         """Sets the accumulator's value; only usable in driver program"""
         if self._deserialized:
             raise PySparkRuntimeError(
-                error_class="VALUE_NOT_ACCESSIBLE",
-                message_parameters={
+                errorClass="VALUE_NOT_ACCESSIBLE",
+                messageParameters={
                     "value": "Accumulator.value",
                 },
             )

python/pyspark/conf.py

@@ -188,8 +188,8 @@ def setExecutorEnv(
         """Set an environment variable to be passed to executors."""
         if (key is not None and pairs is not None) or (key is None and pairs is None):
             raise PySparkRuntimeError(
-                error_class="KEY_VALUE_PAIR_REQUIRED",
-                message_parameters={},
+                errorClass="KEY_VALUE_PAIR_REQUIRED",
+                messageParameters={},
             )
         elif key is not None:
             self.set("spark.executorEnv.{}".format(key), cast(str, value))

python/pyspark/core/broadcast.py

@@ -59,8 +59,8 @@ def _from_id(bid: int) -> "Broadcast[Any]":
 
     if bid not in _broadcastRegistry:
         raise PySparkRuntimeError(
-            error_class="BROADCAST_VARIABLE_NOT_LOADED",
-            message_parameters={
+            errorClass="BROADCAST_VARIABLE_NOT_LOADED",
+            messageParameters={
                 "variable": str(bid),
             },
         )
@@ -299,8 +299,8 @@ def unpersist(self, blocking: bool = False) -> None:
         """
         if self._jbroadcast is None:
             raise PySparkRuntimeError(
-                error_class="INVALID_BROADCAST_OPERATION",
-                message_parameters={"operation": "unpersisted"},
+                errorClass="INVALID_BROADCAST_OPERATION",
+                messageParameters={"operation": "unpersisted"},
             )
         self._jbroadcast.unpersist(blocking)
 
@@ -329,17 +329,17 @@ def destroy(self, blocking: bool = False) -> None:
         """
         if self._jbroadcast is None:
             raise PySparkRuntimeError(
-                error_class="INVALID_BROADCAST_OPERATION",
-                message_parameters={"operation": "destroyed"},
+                errorClass="INVALID_BROADCAST_OPERATION",
+                messageParameters={"operation": "destroyed"},
             )
         self._jbroadcast.destroy(blocking)
         os.unlink(self._path)
 
     def __reduce__(self) -> Tuple[Callable[[int], "Broadcast[T]"], Tuple[int]]:
         if self._jbroadcast is None:
             raise PySparkRuntimeError(
-                error_class="INVALID_BROADCAST_OPERATION",
-                message_parameters={"operation": "serialized"},
+                errorClass="INVALID_BROADCAST_OPERATION",
+                messageParameters={"operation": "serialized"},
             )
         assert self._pickle_registry is not None
         self._pickle_registry.add(self)

python/pyspark/core/context.py

@@ -184,8 +184,8 @@ def __init__(
     ):
         if "SPARK_CONNECT_MODE_ENABLED" in os.environ and "SPARK_LOCAL_REMOTE" not in os.environ:
             raise PySparkRuntimeError(
-                error_class="CONTEXT_UNAVAILABLE_FOR_REMOTE_CLIENT",
-                message_parameters={},
+                errorClass="CONTEXT_UNAVAILABLE_FOR_REMOTE_CLIENT",
+                messageParameters={},
             )
 
         if conf is None or conf.get("spark.executor.allowSparkContext", "false").lower() != "true":
@@ -271,13 +271,13 @@ def _do_init(
         # Check that we have at least the required parameters
         if not self._conf.contains("spark.master"):
             raise PySparkRuntimeError(
-                error_class="MASTER_URL_NOT_SET",
-                message_parameters={},
+                errorClass="MASTER_URL_NOT_SET",
+                messageParameters={},
             )
         if not self._conf.contains("spark.app.name"):
             raise PySparkRuntimeError(
-                error_class="APPLICATION_NAME_NOT_SET",
-                message_parameters={},
+                errorClass="APPLICATION_NAME_NOT_SET",
+                messageParameters={},
             )
 
         # Read back our properties from the conf in case we loaded some of them from
@@ -465,8 +465,8 @@ def _ensure_initialized(
     def __getnewargs__(self) -> NoReturn:
         # This method is called when attempting to pickle SparkContext, which is always an error:
         raise PySparkRuntimeError(
-            error_class="CONTEXT_ONLY_VALID_ON_DRIVER",
-            message_parameters={},
+            errorClass="CONTEXT_ONLY_VALID_ON_DRIVER",
+            messageParameters={},
         )
 
     def __enter__(self) -> "SparkContext":
@@ -2535,8 +2535,8 @@ def show_profiles(self) -> None:
             self.profiler_collector.show_profiles()
         else:
             raise PySparkRuntimeError(
-                error_class="INCORRECT_CONF_FOR_PROFILE",
-                message_parameters={},
+                errorClass="INCORRECT_CONF_FOR_PROFILE",
+                messageParameters={},
             )
 
     def dump_profiles(self, path: str) -> None:
@@ -2552,8 +2552,8 @@ def dump_profiles(self, path: str) -> None:
             self.profiler_collector.dump_profiles(path)
         else:
             raise PySparkRuntimeError(
-                error_class="INCORRECT_CONF_FOR_PROFILE",
-                message_parameters={},
+                errorClass="INCORRECT_CONF_FOR_PROFILE",
+                messageParameters={},
             )
 
     def getConf(self) -> SparkConf:
@@ -2591,8 +2591,8 @@ def _assert_on_driver() -> None:
         """
         if TaskContext.get() is not None:
             raise PySparkRuntimeError(
-                error_class="CONTEXT_ONLY_VALID_ON_DRIVER",
-                message_parameters={},
+                errorClass="CONTEXT_ONLY_VALID_ON_DRIVER",
+                messageParameters={},
             )
 
 

python/pyspark/core/rdd.py

@@ -137,8 +137,8 @@ def portable_hash(x: Hashable) -> int:
 
     if "PYTHONHASHSEED" not in os.environ:
         raise PySparkRuntimeError(
-            error_class="PYTHON_HASH_SEED_NOT_SET",
-            message_parameters={},
+            errorClass="PYTHON_HASH_SEED_NOT_SET",
+            messageParameters={},
         )
 
     if x is None:
@@ -246,8 +246,8 @@ def __repr__(self) -> str:
     def __getnewargs__(self) -> NoReturn:
         # This method is called when attempting to pickle an RDD, which is always an error:
         raise PySparkRuntimeError(
-            error_class="RDD_TRANSFORM_ONLY_VALID_ON_DRIVER",
-            message_parameters={},
+            errorClass="RDD_TRANSFORM_ONLY_VALID_ON_DRIVER",
+            messageParameters={},
         )
 
     @property
@@ -1581,8 +1581,8 @@ def check_return_code() -> Iterable[int]:
                 pipe.wait()
                 if checkCode and pipe.returncode:
                     raise PySparkRuntimeError(
-                        error_class="PIPE_FUNCTION_EXITED",
-                        message_parameters={
+                        errorClass="PIPE_FUNCTION_EXITED",
+                        messageParameters={
                             "func_name": command,
                             "error_code": str(pipe.returncode),
                         },
@@ -5105,8 +5105,8 @@ def toDF(
         self: "RDD[Any]", schema: Optional[Any] = None, sampleRatio: Optional[float] = None
     ) -> "DataFrame":
         raise PySparkRuntimeError(
-            error_class="CALL_BEFORE_INITIALIZE",
-            message_parameters={
+            errorClass="CALL_BEFORE_INITIALIZE",
+            messageParameters={
                 "func_name": "RDD.toDF",
                 "object": "SparkSession",
             },

python/pyspark/errors/exceptions/base.py

@@ -34,24 +34,24 @@ class PySparkException(Exception):
     def __init__(
         self,
         message: Optional[str] = None,
-        error_class: Optional[str] = None,
-        message_parameters: Optional[Dict[str, str]] = None,
-        query_contexts: Optional[List["QueryContext"]] = None,
+        errorClass: Optional[str] = None,
+        messageParameters: Optional[Dict[str, str]] = None,
+        contexts: Optional[List["QueryContext"]] = None,
     ):
-        if query_contexts is None:
-            query_contexts = []
+        if contexts is None:
+            contexts = []
         self._error_reader = ErrorClassesReader()
 
         if message is None:
             self._message = self._error_reader.get_error_message(
-                cast(str, error_class), cast(Dict[str, str], message_parameters)
+                cast(str, errorClass), cast(Dict[str, str], messageParameters)
             )
         else:
             self._message = message
 
-        self._error_class = error_class
-        self._message_parameters = message_parameters
-        self._query_contexts = query_contexts
+        self._errorClass = errorClass
+        self._messageParameters = messageParameters
+        self._contexts = contexts
 
     def getErrorClass(self) -> Optional[str]:
         """
@@ -66,7 +66,7 @@ def getErrorClass(self) -> Optional[str]:
         :meth:`PySparkException.getQueryContext`
         :meth:`PySparkException.getSqlState`
         """
-        return self._error_class
+        return self._errorClass
 
     def getMessageParameters(self) -> Optional[Dict[str, str]]:
         """
@@ -81,7 +81,7 @@ def getMessageParameters(self) -> Optional[Dict[str, str]]:
         :meth:`PySparkException.getQueryContext`
         :meth:`PySparkException.getSqlState`
         """
-        return self._message_parameters
+        return self._messageParameters
 
     def getSqlState(self) -> Optional[str]:
         """
@@ -128,28 +128,28 @@ def getQueryContext(self) -> List["QueryContext"]:
         :meth:`PySparkException.getMessage`
         :meth:`PySparkException.getSqlState`
         """
-        return self._query_contexts
+        return self._contexts
 
     def _log_exception(self) -> None:
-        query_contexts = self.getQueryContext()
-        query_context = query_contexts[0] if len(query_contexts) != 0 else None
-        if query_context:
-            if query_context.contextType().name == "DataFrame":
+        contexts = self.getQueryContext()
+        context = contexts[0] if len(contexts) != 0 else None
+        if context:
+            if context.contextType().name == "DataFrame":
                 logger = PySparkLogger.getLogger("DataFrameQueryContextLogger")
-                call_site = query_context.callSite().split(":")
+                call_site = context.callSite().split(":")
                 line = call_site[1] if len(call_site) == 2 else ""
                 logger.exception(
                     self.getMessage(),
                     file=call_site[0],
                     line=line,
-                    fragment=query_context.fragment(),
-                    error_class=self.getErrorClass(),
+                    fragment=context.fragment(),
+                    errorClass=self.getErrorClass(),
                 )
             else:
                 logger = PySparkLogger.getLogger("SQLQueryContextLogger")
                 logger.exception(
                     self.getMessage(),
-                    error_class=self.getErrorClass(),
+                    errorClass=self.getErrorClass(),
                 )
 
     def __str__(self) -> str:
@@ -299,11 +299,11 @@ class PySparkAssertionError(PySparkException, AssertionError):
     def __init__(
         self,
         message: Optional[str] = None,
-        error_class: Optional[str] = None,
-        message_parameters: Optional[Dict[str, str]] = None,
+        errorClass: Optional[str] = None,
+        messageParameters: Optional[Dict[str, str]] = None,
         data: Optional[Iterable["Row"]] = None,
     ):
-        super().__init__(message, error_class, message_parameters)
+        super().__init__(message, errorClass, messageParameters)
         self.data = data
 
 

python/pyspark/errors/exceptions/captured.py

@@ -146,11 +146,11 @@ def getMessage(self) -> str:
         if self._origin is not None and is_instance_of(
             gw, self._origin, "org.apache.spark.SparkThrowable"
         ):
-            error_class = self._origin.getErrorClass()
-            message_parameters = self._origin.getMessageParameters()
+            errorClass = self._origin.getErrorClass()
+            messageParameters = self._origin.getMessageParameters()
 
             error_message = gw.jvm.org.apache.spark.SparkThrowableHelper.getMessage(
-                error_class, message_parameters
+                errorClass, messageParameters
             )
 
             return error_message