code: def test_split_small(ray_start_regular_shared):
      x = [Counter.remote() for _ in range(4)]
      data = ["a", "b", "c", "d", "e", "f", "g", "h", "i", "j"]
      fail = []
  
      @ray.remote(num_cpus=0)
      def take(s):
          return extract_values("item", s.take())
  
      for m in [1, 3]:
          for n in [1, 3]:
              for locality_hints in [None, x[:n]]:
                  for equal in [True, False]:
                      print("Testing", m, n, equal, locality_hints)
                      ds = ray.data.from_items(data, parallelism=m)
                      splits = ds.split(
                          n, equal=equal, locality_hints=locality_hints
                      )
                      assert len(splits) == n
                      outs = ray.get([take.remote(s) for s in splits])
                      out = []
                      for r in outs:
                          out.extend(r)
                      if equal:
                          lens = set([len(s) for s in outs])  # noqa
                          limit = len(data) - (len(data) % n)
                          allowed = [limit]
                          # Allow for some pipelining artifacts.
                          print(len(out), len(set(out)), allowed)
                          if (
                              len(out) not in allowed
                              or len(set(out)) != len(out)
                              # TODO(ekl) we should be able to enable this check, but
                              # there are some edge condition bugs in split.
                              # or len(lens) != 1
                          ):
                              print("FAIL", m, n, equal, locality_hints)
                              fail.append((m, n, equal, locality_hints))
                      else:
                          if sorted(out) != data:
                              print("FAIL", m, n, equal, locality_hints)
                              fail.append((m, n, equal, locality_hints))
  
      assert not fail, fail
  
imports: import itertools
  import math
  import random
  import time
  from unittest.mock import patch
  import numpy
  import pandas
  import pytest
  import ray
  from ray.data._internal.block_list import BlockList
  from ray.data._internal.equalize import _equalize
  from ray.data._internal.execution.interfaces import RefBundle
  from ray.data._internal.logical.interfaces import LogicalPlan
  from ray.data._internal.logical.operators.input_data_operator import InputData
  from ray.data._internal.plan import ExecutionPlan
  from ray.data._internal.split import _drop_empty_block_split, _generate_global_split_results, _generate_per_block_split_indices, _generate_valid_indices, _split_at_indices, _split_single_block
  from ray.data._internal.stats import DatasetStats
  from ray.data.block import BlockAccessor, BlockMetadata
  from ray.data.dataset import Dataset
  from ray.data.tests.conftest import *
  from ray.data.tests.util import extract_values
  from ray.tests.conftest import *
label_desc: Use image node_id 5 for calling a global function with example usage: test_split_small(ray_start_regular_shared) and returns: extract_values