code: def test_callable_classes(shutdown_only):
      ray.init(num_cpus=2)
      ds = ray.data.range(10, parallelism=10)
  
      class StatefulFn:
          def __init__(self):
              self.num_reuses = 0
  
          def __call__(self, x):
              r = self.num_reuses
              self.num_reuses += 1
              return {"id": np.array([r])}
  
      # map
      actor_reuse = ds.map(StatefulFn, concurrency=1).take()
      assert sorted(extract_values("id", actor_reuse)) == list(
          range(10)
      ), actor_reuse
  
      class StatefulFn:
          def __init__(self):
              self.num_reuses = 0
  
          def __call__(self, x):
              r = self.num_reuses
              self.num_reuses += 1
              return [{"id": r}]
  
      # flat map
      actor_reuse = extract_values(
          "id", ds.flat_map(StatefulFn, concurrency=1).take()
      )
      assert sorted(actor_reuse) == list(range(10)), actor_reuse
  
      class StatefulFn:
          def __init__(self):
              self.num_reuses = 0
  
          def __call__(self, x):
              r = self.num_reuses
              self.num_reuses += 1
              return {"id": np.array([r])}
  
      # map batches
      actor_reuse = extract_values(
          "id",
          ds.map_batches(
              StatefulFn, batch_size=1, concurrency=1
          ).take(),
      )
      assert sorted(actor_reuse) == list(range(10)), actor_reuse
  
      class StatefulFn:
          def __init__(self):
              self.num_reuses = 0
  
          def __call__(self, x):
              r = self.num_reuses
              self.num_reuses += 1
              return r > 0
  
      # filter
      actor_reuse = ds.filter(StatefulFn, concurrency=1).take()
      assert len(actor_reuse) == 9, actor_reuse
  
      class StatefulFnWithArgs:
          def __init__(self, arg, kwarg):
              assert arg == 1
              assert kwarg == 2
  
          def __call__(self, x, arg, kwarg):
              assert arg == 1
              assert kwarg == 2
              return x
  
      # map_batches & map with args & kwargs
      for ds_map in (ds.map_batches, ds.map):
          result = ds_map(
              StatefulFnWithArgs,
              concurrency=1,
              fn_args=(1,),
              fn_kwargs={"kwarg": 2},
              fn_constructor_args=(1,),
              fn_constructor_kwargs={"kwarg": 2},
          ).take()
          assert sorted(extract_values("id", result)) == list(
              range(10)
          ), result
  
      class StatefulFlatMapFnWithArgs:
          def __init__(self, arg, kwarg):
              self._arg = arg
              assert arg == 1
              assert kwarg == 2
  
          def __call__(self, x, arg, kwarg):
              assert arg == 1
              assert kwarg == 2
              return [x] * self._arg
  
      # flat_map with args & kwargs
      result = ds.flat_map(
          StatefulFlatMapFnWithArgs,
          concurrency=1,
          fn_args=(1,),
          fn_kwargs={"kwarg": 2},
          fn_constructor_args=(1,),
          fn_constructor_kwargs={"kwarg": 2},
      ).take()
      assert sorted(extract_values("id", result)) == list(
          range(10)
      ), result
  
imports: import itertools
  import math
  import os
  import signal
  import threading
  import time
  from typing import Iterator
  import numpy
  import pandas
  import pyarrow
  import pyarrow.parquet
  import pytest
  import ray
  from ray._private.test_utils import wait_for_condition
  from ray.data.block import BlockAccessor
  from ray.data.context import DataContext
  from ray.data.tests.conftest import *
  from ray.data.tests.util import column_udf, column_udf_class, extract_values
  from ray.tests.conftest import *
label_desc: Use image node_id 2 for calling a global function with example usage: test_callable_classes(shutdown_only) and returns: dict, list, dict, unknown, x, unknown