code: def wrap_grouped_map_pandas_udf_with_state(f, return_type):
      """
      Provides a new lambda instance wrapping user function of applyInPandasWithState.
  
      The lambda instance receives (key series, iterator of value series, state) and performs
      some conversion to be adapted with the signature of user function.
  
      See the function doc of inner function `wrapped` for more details on what adapter does.
      See the function doc of `mapper` function for
      `eval_type == PythonEvalType.SQL_GROUPED_MAP_PANDAS_UDF_WITH_STATE` for more details on
      the input parameters of lambda function.
  
      Along with the returned iterator, the lambda instance will also produce the return_type as
      converted to the arrow schema.
      """
  
      def wrapped(key_series, value_series_gen, state):
          """
          Provide an adapter of the user function performing below:
  
          - Extract the first value of all columns in key series and produce as a tuple.
          - If the state has timed out, call the user function with empty pandas DataFrame.
          - If not, construct a new generator which converts each element of value series to
            pandas DataFrame (lazy evaluation), and call the user function with the generator
          - Verify each element of returned iterator to check the schema of pandas DataFrame.
          """
          import pandas as pd
  
          key = tuple(s[0] for s in key_series)
  
          if state.hasTimedOut:
              # Timeout processing pass empty iterator. Here we return an empty DataFrame instead.
              values = [
                  pd.DataFrame(
                      columns=pd.concat(
                          next(value_series_gen), axis=1
                      ).columns
                  ),
              ]
          else:
              values = (pd.concat(x, axis=1) for x in value_series_gen)
  
          result_iter = f(key, values, state)
  
          def verify_element(result):
              if not isinstance(result, pd.DataFrame):
                  raise PySparkTypeError(
                      error_class="UDF_RETURN_TYPE",
                      message_parameters={
                          "expected": "iterator of pandas.DataFrame",
                          "actual": "iterator of {}".format(
                              type(result).__name__
                          ),
                      },
                  )
              # the number of columns of result have to match the return type
              # but it is fine for result to have no columns at all if it is empty
              if not (
                  len(result.columns) == len(return_type)
                  or (len(result.columns) == 0 and result.empty)
              ):
                  raise PySparkRuntimeError(
                      error_class="RESULT_LENGTH_MISMATCH_FOR_PANDAS_UDF",
                      message_parameters={
                          "expected": str(len(return_type)),
                          "actual": str(len(result.columns)),
                      },
                  )
  
              return result
  
          if isinstance(result_iter, pd.DataFrame):
              raise PySparkTypeError(
                  error_class="UDF_RETURN_TYPE",
                  message_parameters={
                      "expected": "iterable of pandas.DataFrame",
                      "actual": type(result_iter).__name__,
                  },
              )
  
          try:
              iter(result_iter)
          except TypeError:
              raise PySparkTypeError(
                  error_class="UDF_RETURN_TYPE",
                  message_parameters={
                      "expected": "iterable",
                      "actual": type(result_iter).__name__,
                  },
              )
  
          result_iter_with_validation = (
              verify_element(x) for x in result_iter
          )
  
          return (
              result_iter_with_validation,
              state,
          )
  
      return lambda k, v, s: [
          (wrapped(k, v, s), to_arrow_type(return_type))
      ]
  
imports: import os
  import sys
  import dataclasses
  import time
  import inspect
  import json
  from typing import Any, Callable, Iterable, Iterator, Optional
  import faulthandler
  from pyspark.accumulators import _accumulatorRegistry
  from pyspark.java_gateway import local_connect_and_auth
  from pyspark.taskcontext import BarrierTaskContext, TaskContext
  from pyspark.resource import ResourceInformation
  from pyspark.rdd import PythonEvalType
  from pyspark.serializers import write_int, read_long, read_bool, write_long, read_int, SpecialLengths, UTF8Deserializer, CPickleSerializer, BatchedSerializer
  from pyspark.sql.functions import SkipRestOfInputTableException
  from pyspark.sql.pandas.serializers import ArrowStreamPandasUDFSerializer, ArrowStreamPandasUDTFSerializer, CogroupArrowUDFSerializer, CogroupPandasUDFSerializer, ArrowStreamUDFSerializer, ArrowStreamGroupUDFSerializer, ApplyInPandasWithStateSerializer
  from pyspark.sql.pandas.types import to_arrow_type
  from pyspark.sql.types import ArrayType, BinaryType, DataType, MapType, Row, StringType, StructType, _create_row, _parse_datatype_json_string
  from pyspark.util import fail_on_stopiteration, handle_worker_exception
  from pyspark import shuffle
  from pyspark.errors import PySparkRuntimeError, PySparkTypeError
  from pyspark.worker_util import check_python_version, read_command, pickleSer, send_accumulator_updates, setup_broadcasts, setup_memory_limits, setup_spark_files, utf8_deserializer
label_desc: Use image node_id 14 for calling a global function with example usage: wrap_grouped_map_pandas_udf_with_state(f, return_type) and returns: function, result_iter_with_validation,  state, result