code: def __next__(self):
      if not self._input_blocks_list:
          raise StopIteration
  
      block = self._input_blocks_list.pop(0)
      # NOTE(swang): Results are shuffled between map and merge tasks, so
      # there is no advantage to colocating specific map and merge tasks.
      # Therefore, we do not specify a node affinity policy for map tasks
      # in case the caller or Ray has a better scheduling strategy, e.g.,
      # based on data locality.
      map_result = self._shuffle_map.remote(
          self._mapper_idx,
          block,
          *self._map_args,
      )
      metadata_ref = map_result.pop(-1)
      self._map_results.append(map_result)
      self._mapper_idx += 1
      return metadata_ref
  
imports: import logging
  import math
  from typing import Any, Callable, Dict, List, Optional, Tuple, TypeVar, Union
  import ray
  from ray.data._internal.block_list import BlockList
  from ray.data._internal.execution.interfaces import TaskContext
  from ray.data._internal.progress_bar import ProgressBar
  from ray.data._internal.remote_fn import cached_remote_fn
  from ray.data._internal.shuffle import ShuffleOp
  from ray.data.block import Block, BlockAccessor, BlockExecStats, BlockMetadata
  from ray.data.context import DataContext
  from ray.types import ObjectRef
  from ray.util.scheduling_strategies import NodeAffinitySchedulingStrategy
label_desc: Use image node_id 3 for calling the _MapStageIterator obj's underlying member method code with example usage: obj.__next__() and returns: metadata_ref