code: def test_stats_actor_metrics():
      ray.init(object_store_memory=100e6)
      with patch_update_stats_actor() as update_fn:
          ds = (
              ray.data.range(1000 * 80 * 80 * 4)
              .map_batches(lambda x: x)
              .materialize()
          )
  
      # last emitted metrics from map operator
      final_metric = update_fn.call_args_list[-1].args[1][-1]
  
      assert (
          final_metric.obj_store_mem_spilled
          == ds._plan.stats().dataset_bytes_spilled
      )
      assert (
          final_metric.obj_store_mem_alloc
          == ds._plan.stats().extra_metrics["obj_store_mem_alloc"]
      )
      assert (
          final_metric.obj_store_mem_freed
          == ds._plan.stats().extra_metrics["obj_store_mem_freed"]
      )
      assert (
          final_metric.bytes_outputs_generated == 1000 * 80 * 80 * 4 * 8
      )  # 8B per int
      assert final_metric.rows_outputs_generated == 1000 * 80 * 80 * 4
      # There should be nothing in object store at the end of execution.
      assert final_metric.obj_store_mem_cur == 0
  
      args = update_fn.call_args_list[-1].args
      assert args[0] == f"dataset_{ds._uuid}"
      assert args[2][0] == "Input0"
      assert args[2][1] == "ReadRange->MapBatches(<lambda>)1"
  
      def sleep_three(x):
          import time
  
          time.sleep(3)
          return x
  
      with patch_update_stats_actor() as update_fn:
          ds = (
              ray.data.range(3)
              .map_batches(sleep_three, batch_size=1)
              .materialize()
          )
  
      final_metric = update_fn.call_args_list[-1].args[1][-1]
      assert final_metric.block_generation_time >= 9
  
imports: import re
  import threading
  import time
  from collections import Counter
  from contextlib import contextmanager
  from typing import List, Optional
  from unittest.mock import patch
  import numpy
  import pytest
  import ray
  from ray._private.test_utils import wait_for_condition
  from ray.data._internal.dataset_logger import DatasetLogger
  from ray.data._internal.stats import DatasetStats, StatsManager, _get_or_create_stats_actor, _StatsActor
  from ray.data._internal.util import create_dataset_tag
  from ray.data.block import BlockMetadata
  from ray.data.context import DataContext
  from ray.data.tests.util import column_udf
  from ray.tests.conftest import *
label_desc: Use image node_id 27 for calling a global function with example usage: test_stats_actor_metrics() and returns: x