code: def __next__(self):
      if not self._map_result_buffer or not self._map_result_buffer[0]:
          assert self._merge_idx == 0
          self._map_result_buffer = (
              self._map_stage_iter.pop_map_results()
          )
  
      if not self._map_result_buffer:
          raise StopIteration
  
      # Shuffle the map results for the merge tasks.
      merge_args = [
          map_result.pop(0) for map_result in self._map_result_buffer
      ]
      num_merge_returns = (
          self._stage.merge_schedule.get_num_reducers_per_merge_idx(
              self._merge_idx
          )
      )
      merge_result = self._shuffle_merge.options(
          num_returns=1 + num_merge_returns,
          **self._stage.get_merge_task_options(self._merge_idx),
      ).remote(
          *merge_args,
          reduce_args=self._reduce_args,
      )
      metadata_ref = merge_result.pop(-1)
      self._all_merge_results[self._merge_idx].append(merge_result)
      del merge_result
  
      self._merge_idx += 1
      self._merge_idx %= (
          self._stage.merge_schedule.num_merge_tasks_per_round
      )
      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 2 for calling the _MergeStageIterator obj's underlying member method code with example usage: obj.__next__() and returns: metadata_ref