code: def _generate_transform_fn_for_map_batches(
      fn: UserDefinedFunction,
  ) -> MapTransformCallable[DataBatch, DataBatch]:
      def transform_fn(
          batches: Iterable[DataBatch], _: TaskContext
      ) -> Iterable[DataBatch]:
          for batch in batches:
              try:
                  if (
                      not isinstance(batch, collections.abc.Mapping)
                      and BlockAccessor.for_block(batch).num_rows() == 0
                  ):
                      # For empty input blocks, we directly ouptut them without
                      # calling the UDF.
                      # TODO(hchen): This workaround is because some all-to-all
                      # operators output empty blocks with no schema.
                      res = [batch]
                  else:
                      res = fn(batch)
                      if not isinstance(res, GeneratorType):
                          res = [res]
              except ValueError as e:
                  read_only_msgs = [
                      "assignment destination is read-only",
                      "buffer source array is read-only",
                  ]
                  err_msg = str(e)
                  if any(msg in err_msg for msg in read_only_msgs):
                      raise ValueError(
                          f"Batch mapper function {fn.__name__} tried to mutate a "
                          "zero-copy read-only batch. To be able to mutate the "
                          "batch, pass zero_copy_batch=False to map_batches(); "
                          "this will create a writable copy of the batch before "
                          "giving it to fn. To elide this copy, modify your mapper "
                          "function so it doesn't try to mutate its input."
                      ) from e
                  else:
                      raise e from None
              else:
                  for out_batch in res:
                      _validate_batch_output(out_batch)
                      yield out_batch
  
      return transform_fn
  
imports: import collections
  from types import GeneratorType
  from typing import Any, Callable, Iterable, Iterator, Optional
  import numpy
  import pandas
  import pyarrow
  import ray
  from ray.data._internal.compute import get_compute
  from ray.data._internal.execution.interfaces import PhysicalOperator
  from ray.data._internal.execution.interfaces.task_context import TaskContext
  from ray.data._internal.execution.operators.map_operator import MapOperator
  from ray.data._internal.execution.operators.map_transformer import BatchMapTransformFn, BlocksToBatchesMapTransformFn, BlocksToRowsMapTransformFn, BuildOutputBlocksMapTransformFn, MapTransformCallable, MapTransformer, Row, RowMapTransformFn
  from ray.data._internal.execution.util import make_callable_class_concurrent
  from ray.data._internal.logical.operators.map_operator import AbstractUDFMap, Filter, FlatMap, MapBatches, MapRows
  from ray.data._internal.numpy_support import is_valid_udf_return
  from ray.data._internal.util import _truncated_repr
  from ray.data.block import Block, BlockAccessor, CallableClass, DataBatch, UserDefinedFunction
  from ray.data.context import DataContext
label_desc: Use image node_id 4 for calling a global function with example usage: _generate_transform_fn_for_map_batches(fn) and returns: transform_fn