code: def sort_columns(  # noqa: C901,PLR0915
      df: DataFrame,
      sort_order: str,
      sort_nested: bool = False,
  ) -> DataFrame:
      """This function sorts the columns of a given DataFrame based on a given sort
      order. The ``sort_order`` parameter can either be ``asc`` or ``desc``, which correspond to
      ascending and descending order, respectively. If any other value is provided for
      the ``sort_order`` parameter, a ``ValueError`` will be raised.
  
      :param df: A DataFrame
      :type df: pyspark.sql.DataFrame
      :param sort_order: The order in which to sort the columns in the DataFrame
      :type sort_order: str
      :param sort_nested: Whether to sort nested structs or not. Defaults to false.
      :type sort_nested: bool
      :return: A DataFrame with the columns sorted in the chosen order
      :rtype: pyspark.sql.DataFrame
      """
  
      def sort_nested_cols(
          schema, is_reversed, base_field=""
      ) -> list[str]:  # noqa: ANN001
          # recursively check nested fields and sort them
          # https://stackoverflow.com/questions/57821538/how-to-sort-columns-of-nested-structs-alphabetically-in-pyspark
          # Credits: @pault for logic
  
          def parse_fields(
              fields_to_sort: list,
              parent_struct,
              is_reversed: bool,  # noqa: ANN001
          ) -> list:
              sorted_fields: list = sorted(
                  fields_to_sort,
                  key=lambda x: x["name"],
                  reverse=is_reversed,
              )
  
              results = []
              for field in sorted_fields:
                  new_struct = StructType([StructField.fromJson(field)])
                  new_base_field = parent_struct.name
                  if base_field:
                      new_base_field = base_field + "." + new_base_field
  
                  results.extend(
                      sort_nested_cols(
                          new_struct,
                          is_reversed,
                          base_field=new_base_field,
                      ),
                  )
              return results
  
          select_cols = []
          for parent_struct in sorted(
              schema, key=lambda x: x.name, reverse=is_reversed
          ):
              field_type = parent_struct.dataType
              if isinstance(field_type, ArrayType):
                  array_parent = parent_struct.jsonValue()["type"][
                      "elementType"
                  ]
                  base_str = f"transform({parent_struct.name}"
                  suffix_str = f") AS {parent_struct.name}"
  
                  # if struct in array, create mapping to struct
                  if array_parent["type"] == "struct":
                      array_parent = array_parent["fields"]
                      base_str = f"{base_str}, x -> struct("
                      suffix_str = f"){suffix_str}"
  
                  array_elements = parse_fields(
                      array_parent, parent_struct, is_reversed
                  )
                  element_names = [
                      i.split(".")[-1] for i in array_elements
                  ]
                  array_elements_formatted = [
                      f"x.{i} as {i}" for i in element_names
                  ]
  
                  # create a string representation of the sorted array
                  # ex: transform(phone_numbers, x -> struct(x.number as number, x.type as type)) AS phone_numbers
                  result = f"{base_str}{', '.join(array_elements_formatted)}{suffix_str}"
  
              elif isinstance(field_type, StructType):
                  field_list = parent_struct.jsonValue()["type"][
                      "fields"
                  ]
                  sub_fields = parse_fields(
                      field_list, parent_struct, is_reversed
                  )
  
                  # create a string representation of the sorted struct
                  # ex: struct(address.zip.first5, address.zip.last4) AS zip
                  result = f"struct({', '.join(sub_fields)}) AS {parent_struct.name}"
  
              elif base_field:
                  result = f"{base_field}.{parent_struct.name}"
              else:
                  result = parent_struct.name
              select_cols.append(result)
  
          return select_cols
  
      def get_original_nullability(
          field: StructField, result_dict: dict
      ) -> None:
          if hasattr(field, "nullable"):
              result_dict[field.name] = field.nullable
          else:
              result_dict[field.name] = True
  
          if not isinstance(
              field.dataType, StructType
          ) and not isinstance(
              field.dataType,
              ArrayType,
          ):
              return
  
          if isinstance(field.dataType, ArrayType):
              result_dict[
                  f"{field.name}_element"
              ] = field.dataType.containsNull
              children = field.dataType.elementType.fields
          else:
              children = field.dataType.fields
          for i in children:
              get_original_nullability(i, result_dict)
  
      def fix_nullability(
          field: StructField, result_dict: dict
      ) -> None:
          field.nullable = result_dict[field.name]
          if not isinstance(
              field.dataType, StructType
          ) and not isinstance(
              field.dataType,
              ArrayType,
          ):
              return
  
          if isinstance(field.dataType, ArrayType):
              # save the containsNull property of the ArrayType
              field.dataType.containsNull = result_dict[
                  f"{field.name}_element"
              ]
              children = field.dataType.elementType.fields
          else:
              children = field.dataType.fields
  
          for i in children:
              fix_nullability(i, result_dict)
  
      if sort_order not in ["asc", "desc"]:
          msg = f"['asc', 'desc'] are the only valid sort orders and you entered a sort order of '{sort_order}'"
          raise ValueError(
              msg,
          )
      reverse_lookup = {
          "asc": False,
          "desc": True,
      }
  
      is_reversed: bool = reverse_lookup[sort_order]
      top_level_sorted_df = df.select(
          *sorted(df.columns, reverse=is_reversed)
      )
      if not sort_nested:
          return top_level_sorted_df
  
      is_nested: bool = any(
          isinstance(i.dataType, (StructType, ArrayType))
          for i in top_level_sorted_df.schema
      )
  
      if not is_nested:
          return top_level_sorted_df
  
      fully_sorted_schema = sort_nested_cols(
          top_level_sorted_df.schema, is_reversed
      )
      output = df.selectExpr(fully_sorted_schema)
      result_dict = {}
      for field in df.schema:
          get_original_nullability(field, result_dict)
  
      for field in output.schema:
          fix_nullability(field, result_dict)
  
      if not hasattr(SparkSession, "getActiveSession"):  # spark 2.4
          spark = SparkSession.builder.getOrCreate()
      else:
          spark = SparkSession.getActiveSession()
          spark = (
              spark
              if spark is not None
              else SparkSession.builder.getOrCreate()
          )
  
      return spark.createDataFrame(output.rdd, output.schema)
  
file_path: quinn/quinn/transformations.py
imports: from __future__ import annotations
  import re
  from collections.abc import Callable
  from pyspark.sql import DataFrame, SparkSession
  from pyspark.sql import functions
  from pyspark.sql.types import ArrayType, MapType, StructField, StructType
  from quinn.schema_helpers import complex_fields
label_desc: Use image node_id 5 for calling a global function with example usage: sort_columns(df, sort_order, sort_nested) and returns: spark, select_cols, top_level_sorted_df, top_level_sorted_df, results