code: def convert_exception(
      info: "ErrorInfo",
      truncated_message: str,
      resp: Optional[pb2.FetchErrorDetailsResponse],
      display_server_stacktrace: bool = False,
  ) -> SparkConnectException:
      classes = []
      sql_state = None
      error_class = None
      message_parameters = None
      query_contexts: Optional[List[BaseQueryContext]] = None
  
      if "classes" in info.metadata:
          classes = json.loads(info.metadata["classes"])
  
      if "sqlState" in info.metadata:
          sql_state = info.metadata["sqlState"]
  
      if "errorClass" in info.metadata:
          error_class = info.metadata["errorClass"]
  
      if "messageParameters" in info.metadata:
          message_parameters = json.loads(
              info.metadata["messageParameters"]
          )
  
      stacktrace: Optional[str] = None
      if resp is not None and resp.HasField("root_error_idx"):
          message = resp.errors[resp.root_error_idx].message
          stacktrace = _extract_jvm_stacktrace(resp)
      else:
          message = truncated_message
          stacktrace = (
              info.metadata["stackTrace"]
              if "stackTrace" in info.metadata
              else None
          )
          display_server_stacktrace = (
              display_server_stacktrace
              if stacktrace is not None
              else False
          )
  
      if (
          resp is not None
          and resp.errors
          and hasattr(
              resp.errors[resp.root_error_idx], "spark_throwable"
          )
      ):
          message_parameters = dict(
              resp.errors[
                  resp.root_error_idx
              ].spark_throwable.message_parameters
          )
          query_contexts = []
          for query_context in resp.errors[
              resp.root_error_idx
          ].spark_throwable.query_contexts:
              query_contexts.append(QueryContext(query_context))
  
      if (
          "org.apache.spark.sql.catalyst.parser.ParseException"
          in classes
      ):
          return ParseException(
              message,
              error_class=error_class,
              message_parameters=message_parameters,
              sql_state=sql_state,
              server_stacktrace=stacktrace,
              display_server_stacktrace=display_server_stacktrace,
              query_contexts=query_contexts,
          )
      # Order matters. ParseException inherits AnalysisException.
      elif "org.apache.spark.sql.AnalysisException" in classes:
          return AnalysisException(
              message,
              error_class=error_class,
              message_parameters=message_parameters,
              sql_state=sql_state,
              server_stacktrace=stacktrace,
              display_server_stacktrace=display_server_stacktrace,
              query_contexts=query_contexts,
          )
      elif (
          "org.apache.spark.sql.streaming.StreamingQueryException"
          in classes
      ):
          return StreamingQueryException(
              message,
              error_class=error_class,
              message_parameters=message_parameters,
              sql_state=sql_state,
              server_stacktrace=stacktrace,
              display_server_stacktrace=display_server_stacktrace,
              query_contexts=query_contexts,
          )
      elif (
          "org.apache.spark.sql.execution.QueryExecutionException"
          in classes
      ):
          return QueryExecutionException(
              message,
              error_class=error_class,
              message_parameters=message_parameters,
              sql_state=sql_state,
              server_stacktrace=stacktrace,
              display_server_stacktrace=display_server_stacktrace,
              query_contexts=query_contexts,
          )
      # Order matters. NumberFormatException inherits IllegalArgumentException.
      elif "java.lang.NumberFormatException" in classes:
          return NumberFormatException(
              message,
              error_class=error_class,
              message_parameters=message_parameters,
              sql_state=sql_state,
              server_stacktrace=stacktrace,
              display_server_stacktrace=display_server_stacktrace,
              query_contexts=query_contexts,
          )
      elif "java.lang.IllegalArgumentException" in classes:
          return IllegalArgumentException(
              message,
              error_class=error_class,
              message_parameters=message_parameters,
              sql_state=sql_state,
              server_stacktrace=stacktrace,
              display_server_stacktrace=display_server_stacktrace,
              query_contexts=query_contexts,
          )
      elif "java.lang.ArithmeticException" in classes:
          return ArithmeticException(
              message,
              error_class=error_class,
              message_parameters=message_parameters,
              sql_state=sql_state,
              server_stacktrace=stacktrace,
              display_server_stacktrace=display_server_stacktrace,
              query_contexts=query_contexts,
          )
      elif "java.lang.UnsupportedOperationException" in classes:
          return UnsupportedOperationException(
              message,
              error_class=error_class,
              message_parameters=message_parameters,
              sql_state=sql_state,
              server_stacktrace=stacktrace,
              display_server_stacktrace=display_server_stacktrace,
              query_contexts=query_contexts,
          )
      elif "java.lang.ArrayIndexOutOfBoundsException" in classes:
          return ArrayIndexOutOfBoundsException(
              message,
              error_class=error_class,
              message_parameters=message_parameters,
              sql_state=sql_state,
              server_stacktrace=stacktrace,
              display_server_stacktrace=display_server_stacktrace,
              query_contexts=query_contexts,
          )
      elif "java.time.DateTimeException" in classes:
          return DateTimeException(
              message,
              error_class=error_class,
              sql_state=sql_state,
              server_stacktrace=stacktrace,
              display_server_stacktrace=display_server_stacktrace,
              query_contexts=query_contexts,
          )
      elif "org.apache.spark.SparkRuntimeException" in classes:
          return SparkRuntimeException(
              message,
              error_class=error_class,
              message_parameters=message_parameters,
              sql_state=sql_state,
              server_stacktrace=stacktrace,
              display_server_stacktrace=display_server_stacktrace,
              query_contexts=query_contexts,
          )
      elif "org.apache.spark.SparkUpgradeException" in classes:
          return SparkUpgradeException(
              message,
              error_class=error_class,
              message_parameters=message_parameters,
              sql_state=sql_state,
              server_stacktrace=stacktrace,
              display_server_stacktrace=display_server_stacktrace,
              query_contexts=query_contexts,
          )
      elif "org.apache.spark.api.python.PythonException" in classes:
          return PythonException(
              "\n  An exception was thrown from the Python worker. "
              "Please see the stack trace below.\n%s" % message
          )
      elif "org.apache.spark.SparkNoSuchElementException" in classes:
          return SparkNoSuchElementException(
              message,
              error_class=error_class,
              message_parameters=message_parameters,
              sql_state=sql_state,
              server_stacktrace=stacktrace,
              display_server_stacktrace=display_server_stacktrace,
              query_contexts=query_contexts,
          )
      # Make sure that the generic SparkException is handled last.
      elif "org.apache.spark.SparkException" in classes:
          return SparkException(
              message,
              error_class=error_class,
              message_parameters=message_parameters,
              sql_state=sql_state,
              server_stacktrace=stacktrace,
              display_server_stacktrace=display_server_stacktrace,
              query_contexts=query_contexts,
          )
      else:
          return SparkConnectGrpcException(
              message,
              reason=info.reason,
              message_parameters=message_parameters,
              error_class=error_class,
              sql_state=sql_state,
              server_stacktrace=stacktrace,
              display_server_stacktrace=display_server_stacktrace,
              query_contexts=query_contexts,
          )
  
imports: import pyspark.sql.connect.proto
  import json
  from typing import Dict, List, Optional, TYPE_CHECKING
  from pyspark.errors.exceptions.base import AnalysisException, IllegalArgumentException, ArithmeticException, UnsupportedOperationException, ArrayIndexOutOfBoundsException, DateTimeException, NumberFormatException, ParseException, PySparkException, PythonException, StreamingQueryException, QueryExecutionException, SparkRuntimeException, SparkNoSuchElementException, SparkUpgradeException, QueryContext, QueryContextType
label_desc: Use image node_id 1 for calling a global function with example usage: convert_exception(info, truncated_message, resp, display_server_stacktrace) and returns: ParseException, AnalysisException, StreamingQueryException, QueryExecutionException, NumberFormatException, IllegalArgumentException, ArithmeticException, UnsupportedOperationException, ArrayIndexOutOfBoundsException, DateTimeException, SparkRuntimeException, SparkUpgradeException, PythonException, SparkNoSuchElementException, SparkException, SparkConnectGrpcException