code: def pyspark_types_to_proto_types(data_type: DataType) -> pb2.DataType:
      ret = pb2.DataType()
      if isinstance(data_type, NullType):
          ret.null.CopyFrom(pb2.DataType.NULL())
      elif isinstance(data_type, StringType):
          ret.string.CopyFrom(pb2.DataType.String())
      elif isinstance(data_type, BooleanType):
          ret.boolean.CopyFrom(pb2.DataType.Boolean())
      elif isinstance(data_type, BinaryType):
          ret.binary.CopyFrom(pb2.DataType.Binary())
      elif isinstance(data_type, ByteType):
          ret.byte.CopyFrom(pb2.DataType.Byte())
      elif isinstance(data_type, ShortType):
          ret.short.CopyFrom(pb2.DataType.Short())
      elif isinstance(data_type, IntegerType):
          ret.integer.CopyFrom(pb2.DataType.Integer())
      elif isinstance(data_type, LongType):
          ret.long.CopyFrom(pb2.DataType.Long())
      elif isinstance(data_type, FloatType):
          ret.float.CopyFrom(pb2.DataType.Float())
      elif isinstance(data_type, DoubleType):
          ret.double.CopyFrom(pb2.DataType.Double())
      elif isinstance(data_type, DecimalType):
          ret.decimal.scale = data_type.scale
          ret.decimal.precision = data_type.precision
      elif isinstance(data_type, DateType):
          ret.date.CopyFrom(pb2.DataType.Date())
      elif isinstance(data_type, TimestampType):
          ret.timestamp.CopyFrom(pb2.DataType.Timestamp())
      elif isinstance(data_type, TimestampNTZType):
          ret.timestamp_ntz.CopyFrom(pb2.DataType.TimestampNTZ())
      elif isinstance(data_type, DayTimeIntervalType):
          ret.day_time_interval.start_field = data_type.startField
          ret.day_time_interval.end_field = data_type.endField
      elif isinstance(data_type, YearMonthIntervalType):
          ret.year_month_interval.start_field = data_type.startField
          ret.year_month_interval.end_field = data_type.endField
      elif isinstance(data_type, CalendarIntervalType):
          ret.calendar_interval.CopyFrom(
              pb2.DataType.CalendarInterval()
          )
      elif isinstance(data_type, StructType):
          struct = pb2.DataType.Struct()
          for field in data_type.fields:
              struct_field = pb2.DataType.StructField()
              struct_field.name = field.name
              struct_field.data_type.CopyFrom(
                  pyspark_types_to_proto_types(field.dataType)
              )
              struct_field.nullable = field.nullable
              if field.metadata is not None and len(field.metadata) > 0:
                  struct_field.metadata = json.dumps(field.metadata)
              struct.fields.append(struct_field)
          ret.struct.CopyFrom(struct)
      elif isinstance(data_type, MapType):
          ret.map.key_type.CopyFrom(
              pyspark_types_to_proto_types(data_type.keyType)
          )
          ret.map.value_type.CopyFrom(
              pyspark_types_to_proto_types(data_type.valueType)
          )
          ret.map.value_contains_null = data_type.valueContainsNull
      elif isinstance(data_type, ArrayType):
          ret.array.element_type.CopyFrom(
              pyspark_types_to_proto_types(data_type.elementType)
          )
          ret.array.contains_null = data_type.containsNull
      elif isinstance(data_type, UserDefinedType):
          json_value = data_type.jsonValue()
          ret.udt.type = "udt"
          if "class" in json_value:
              # Scala/Java UDT
              ret.udt.jvm_class = json_value["class"]
          else:
              # Python UDT
              ret.udt.serialized_python_class = json_value[
                  "serializedClass"
              ]
          ret.udt.python_class = json_value["pyClass"]
          ret.udt.sql_type.CopyFrom(
              pyspark_types_to_proto_types(data_type.sqlType())
          )
      elif isinstance(data_type, UnparsedDataType):
          data_type_string = data_type.data_type_string
          ret.unparsed.data_type_string = data_type_string
      else:
          raise PySparkValueError(
              error_class="UNSUPPORTED_OPERATION",
              message_parameters={
                  "operation": f"data type {data_type}"
              },
          )
      return ret
  
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 1 for calling a global function with example usage: pyspark_types_to_proto_types(data_type) and returns: ret