code: def _create_log_reg_state(
      spark: SparkSession,
      phenotype_df: pd.DataFrame,
      offset_df: pd.DataFrame,
      sql_type: DataType,
      C: NDArray[(Any, Any), Float],
      correction: str,
      add_intercept: bool,
      contigs: Optional[List[str]],
  ) -> Union[LogRegState, Dict[str, LogRegState]]:
      """
      Creates the broadcasted LogRegState object (or one object per contig if LOCO offsets were provided).
  
      Fitting the null logistic models can be expensive, so the work is distributed across the cluster
      using Pandas UDFs.
      """
      offset_type = gwas_fx._validate_offset(phenotype_df, offset_df)
      if (
          offset_type == gwas_fx._OffsetType.LOCO_OFFSET
          and contigs is not None
      ):
          offset_df = offset_df.loc[pd.IndexSlice[:, contigs], :]
      pivoted_phenotype_df = reshape_for_gwas(spark, phenotype_df)
      result_fields = [
          StructField("label", StringType()),
          StructField("y_res", ArrayType(sql_type)),
          StructField("gamma", ArrayType(sql_type)),
          StructField("inv_CtGammaC", ArrayType(sql_type)),
      ]
  
      if correction == correction_approx_firth:
          result_fields.append(
              StructField("firth_offset", ArrayType(sql_type))
          )
  
      if offset_type == gwas_fx._OffsetType.NO_OFFSET:
          df = pivoted_phenotype_df
      else:
          pivoted_offset_df = reshape_for_gwas(
              spark, offset_df
          ).withColumnRenamed("values", "offset")
          df = pivoted_offset_df.join(pivoted_phenotype_df, on="label")
  
      if offset_type == gwas_fx._OffsetType.LOCO_OFFSET:
          result_fields.append(StructField("contigName", StringType()))
  
      def map_func(pdf_iterator):
          for pdf in pdf_iterator:
              yield pdf.apply(
                  lambda r: _prepare_one_phenotype(
                      C, r, correction, add_intercept
                  ),
                  axis="columns",
                  result_type="expand",
              )
  
      pdf = df.mapInPandas(
          map_func, StructType(result_fields)
      ).toPandas()
      phenotypes = phenotype_df.columns.to_series().astype("str")
      n_covar = C.shape[1]
  
      if offset_type != gwas_fx._OffsetType.LOCO_OFFSET:
          state = _pdf_to_log_reg_state(pdf, phenotypes, n_covar)
          return state
  
      all_contigs = pdf["contigName"].unique()
      return {
          contig: _pdf_to_log_reg_state(
              pdf.loc[pdf.contigName == contig, :], phenotypes, n_covar
          )
          for contig in all_contigs
      }
  
imports: from pyspark.sql.types import ArrayType, BooleanType, StringType, StructField, DataType, StructType
  from typing import Any, List, Optional, Dict, Union
  import pandas
  import numpy
  from pyspark.sql import DataFrame, SparkSession
  import statsmodels.api
  from dataclasses import dataclass
  from typeguard import typechecked
  from nptyping import Float, NDArray, Int32
  from scipy import stats
  import opt_einsum
  from .None import functions
  from .None import approx_firth
  from .functions import _VALUES_COLUMN_NAME, _get_indices_to_drop
  from ..wgr.wgr_functions import reshape_for_gwas
label_desc: Use image node_id 5 for calling a global function with example usage: _create_log_reg_state(spark, phenotype_df, offset_df, sql_type, C, correction, add_intercept, contigs) and returns: unknown, state