code: def test_stats_manager(shutdown_only):
      ray.init()
      num_threads = 10
  
      datasets = [None] * num_threads
      # Mock clear methods so that _last_execution_stats and _last_iteration_stats
      # are not cleared. We will assert on them afterwards.
      with patch.object(
          StatsManager, "clear_execution_metrics"
      ), patch.object(StatsManager, "clear_iteration_metrics"):
  
          def update_stats_manager(i):
              datasets[i] = ray.data.range(10).map_batches(lambda x: x)
              for _ in datasets[i].iter_batches(batch_size=1):
                  pass
  
          threads = [
              threading.Thread(
                  target=update_stats_manager, args=(i,), daemon=True
              )
              for i in range(num_threads)
          ]
          for thread in threads:
              thread.start()
          for thread in threads:
              thread.join()
  
      assert len(StatsManager._last_execution_stats) == num_threads
      assert len(StatsManager._last_iteration_stats) == num_threads
  
      # Clear dataset tags manually.
      for dataset in datasets:
          dataset_tag = create_dataset_tag(dataset._name, dataset._uuid)
          assert dataset_tag in StatsManager._last_execution_stats
          assert dataset_tag in StatsManager._last_iteration_stats
          StatsManager.clear_execution_metrics(
              dataset_tag,
              ["Input0", "ReadRange->MapBatches(<lambda>)1"],
          )
          StatsManager.clear_iteration_metrics(dataset_tag)
  
      wait_for_condition(
          lambda: not StatsManager._update_thread.is_alive()
      )
      prev_thread = StatsManager._update_thread
  
      ray.data.range(100).map_batches(lambda x: x).materialize()
      # Check that a new different thread is spawned.
      assert StatsManager._update_thread != prev_thread
      wait_for_condition(
          lambda: not StatsManager._update_thread.is_alive()
      )
  
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 33 for calling a global function with example usage: test_stats_manager(shutdown_only) without return types