code: def proto_schema_to_pyspark_data_type(
      schema: pb2.DataType,
  ) -> DataType:
      if schema.HasField("null"):
          return NullType()
      elif schema.HasField("boolean"):
          return BooleanType()
      elif schema.HasField("binary"):
          return BinaryType()
      elif schema.HasField("byte"):
          return ByteType()
      elif schema.HasField("short"):
          return ShortType()
      elif schema.HasField("integer"):
          return IntegerType()
      elif schema.HasField("long"):
          return LongType()
      elif schema.HasField("float"):
          return FloatType()
      elif schema.HasField("double"):
          return DoubleType()
      elif schema.HasField("decimal"):
          p = (
              schema.decimal.precision
              if schema.decimal.HasField("precision")
              else 10
          )
          s = (
              schema.decimal.scale
              if schema.decimal.HasField("scale")
              else 0
          )
          return DecimalType(precision=p, scale=s)
      elif schema.HasField("string"):
          return StringType()
      elif schema.HasField("char"):
          return CharType(schema.char.length)
      elif schema.HasField("var_char"):
          return VarcharType(schema.var_char.length)
      elif schema.HasField("date"):
          return DateType()
      elif schema.HasField("timestamp"):
          return TimestampType()
      elif schema.HasField("timestamp_ntz"):
          return TimestampNTZType()
      elif schema.HasField("day_time_interval"):
          start: Optional[int] = (
              schema.day_time_interval.start_field
              if schema.day_time_interval.HasField("start_field")
              else None
          )
          end: Optional[int] = (
              schema.day_time_interval.end_field
              if schema.day_time_interval.HasField("end_field")
              else None
          )
          return DayTimeIntervalType(startField=start, endField=end)
      elif schema.HasField("year_month_interval"):
          start: Optional[int] = (  # type: ignore[no-redef]
              schema.year_month_interval.start_field
              if schema.year_month_interval.HasField("start_field")
              else None
          )
          end: Optional[int] = (  # type: ignore[no-redef]
              schema.year_month_interval.end_field
              if schema.year_month_interval.HasField("end_field")
              else None
          )
          return YearMonthIntervalType(startField=start, endField=end)
      elif schema.HasField("calendar_interval"):
          return CalendarIntervalType()
      elif schema.HasField("array"):
          return ArrayType(
              proto_schema_to_pyspark_data_type(
                  schema.array.element_type
              ),
              schema.array.contains_null,
          )
      elif schema.HasField("struct"):
          fields = []
          for f in schema.struct.fields:
              if f.HasField("metadata"):
                  metadata = json.loads(f.metadata)
              else:
                  metadata = None
              fields.append(
                  StructField(
                      f.name,
                      proto_schema_to_pyspark_data_type(f.data_type),
                      f.nullable,
                      metadata,
                  )
              )
          return StructType(fields)
      elif schema.HasField("map"):
          return MapType(
              proto_schema_to_pyspark_data_type(schema.map.key_type),
              proto_schema_to_pyspark_data_type(schema.map.value_type),
              schema.map.value_contains_null,
          )
      elif schema.HasField("udt"):
          assert schema.udt.type == "udt"
          json_value = {}
          if schema.udt.HasField("python_class"):
              json_value["pyClass"] = schema.udt.python_class
          if schema.udt.HasField("serialized_python_class"):
              json_value[
                  "serializedClass"
              ] = schema.udt.serialized_python_class
          return UserDefinedType.fromJson(json_value)
      else:
          raise PySparkValueError(
              error_class="UNSUPPORTED_OPERATION",
              message_parameters={"operation": f"data type {schema}"},
          )
  
file_path: spark/python/pyspark/sql/connect/types.py
imports: from pyspark.sql.connect.utils import check_dependencies
  import json
  from typing import Any, Dict, Optional
  from pyspark.sql.types import DataType, ByteType, ShortType, IntegerType, FloatType, DateType, TimestampType, TimestampNTZType, DayTimeIntervalType, YearMonthIntervalType, CalendarIntervalType, MapType, StringType, CharType, VarcharType, StructType, StructField, ArrayType, DoubleType, LongType, DecimalType, BinaryType, BooleanType, NullType, UserDefinedType
  from pyspark.errors import PySparkAssertionError, PySparkValueError
  import pyspark.sql.connect.proto
label_desc: Use image node_id 2 for calling a global function with example usage: proto_schema_to_pyspark_data_type(schema) and returns: NullType, BooleanType, BinaryType, ByteType, ShortType, IntegerType, LongType, FloatType, DoubleType, DecimalType, StringType, CharType, VarcharType, DateType, TimestampType, TimestampNTZType, DayTimeIntervalType, YearMonthIntervalType, CalendarIntervalType, ArrayType, StructType, MapType, UserDefinedType