code: def test_spilled_stats(shutdown_only):
      # The object store is about 100MB.
      ray.init(object_store_memory=100e6)
      # The size of dataset is 1000*80*80*4*8B, about 200MB.
      ds = (
          ray.data.range(1000 * 80 * 80 * 4)
          .map_batches(lambda x: x)
          .materialize()
      )
  
      assert (
          canonicalize(ds.stats(), filter_global_stats=False)
          == f"""Operator N ReadRange->MapBatches(<lambda>): {EXECUTION_STRING}
  * Remote wall time: T min, T max, T mean, T total
  * Remote cpu time: T min, T max, T mean, T total
  * Peak heap memory usage (MiB): N min, N max, N mean
  * Output num rows per block: N min, N max, N mean, N total
  * Output size bytes per block: N min, N max, N mean, N total
  * Output rows per task: N min, N max, N mean, N tasks used
  * Tasks per node: N min, N max, N mean; N nodes used
  * Extra metrics: {MEM_SPILLED_EXTRA_METRICS}
  
  Cluster memory:
  * Spilled to disk: M
  * Restored from disk: M
  
  Dataset memory:
  * Spilled to disk: M
  """
      )
  
      # Around 100MB should be spilled (200MB - 100MB)
      assert ds._plan.stats().dataset_bytes_spilled > 100e6
  
      ds = (
          ray.data.range(1000 * 80 * 80 * 4)
          .map_batches(lambda x: x)
          .materialize()
          .map_batches(lambda x: x)
          .materialize()
      )
      # two map_batches operators, twice the spillage
      assert ds._plan.stats().dataset_bytes_spilled > 200e6
  
      # The size of dataset is around 50MB, there should be no spillage
      ds = (
          ray.data.range(250 * 80 * 80 * 4, parallelism=1)
          .map_batches(lambda x: x)
          .materialize()
      )
  
      assert ds._plan.stats().dataset_bytes_spilled == 0
  
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 26 for calling a global function with example usage: test_spilled_stats(shutdown_only) without return types