code: def resolve_block_refs(
      block_ref_iter: Iterator[ObjectRef[Block]],
      stats: Optional[DatasetStats] = None,
  ) -> Iterator[Block]:
      """Resolves the block references for each logical batch.
  
      Args:
          block_ref_iter: An iterator over block object references.
          stats: An optional stats object to recording block hits and misses.
      """
      hits = 0
      misses = 0
      unknowns = 0
  
      for block_ref in block_ref_iter:
          (
              current_hit,
              current_miss,
              current_unknown,
          ) = _calculate_ref_hits([block_ref])
          hits += current_hit
          misses += current_miss
          unknowns += current_unknown
  
          # TODO(amogkam): Optimized further by batching multiple references in a single
          # `ray.get()` call.
          with stats.iter_get_s.timer() if stats else nullcontext():
              block = ray.get(block_ref)
          yield block
  
      if stats:
          stats.iter_blocks_local = hits
          stats.iter_blocks_remote = misses
          stats.iter_unknown_location = unknowns
  
imports: import logging
  import threading
  from contextlib import nullcontext
  from typing import Any, Callable, Iterator, List, Optional, Tuple
  import ray
  from ray.actor import ActorHandle
  from ray.data._internal.batcher import Batcher, ShufflingBatcher
  from ray.data._internal.block_batching.interfaces import Batch, BlockPrefetcher, CollatedBatch
  from ray.data._internal.stats import DatasetStats
  from ray.data.block import Block, BlockAccessor, DataBatch
  from ray.types import ObjectRef
  from ray.util.scheduling_strategies import NodeAffinitySchedulingStrategy
label_desc: Use image node_id 2 for calling a global function with example usage: resolve_block_refs(block_ref_iter, stats) without return types