code: def test_split_hints(ray_start_regular_shared):
      @ray.remote
      class Actor(object):
          def __init__(self):
              pass
  
      def assert_split_assignment(
          block_node_ids, actor_node_ids, expected_split_result
      ):
          """Helper function to setup split hints test.
  
          Args:
              block_node_ids: a list of blocks with their locations. For
                  example ["node1", "node2"] represents two blocks with
                  "node1", "node2" as their location respectively.
              actor_node_ids: a list of actors with their locations. For
                  example ["node1", "node2"] represents two actors with
                  "node1", "node2" as their location respectively.
              expected_split_result: a list of allocation result, each entry
                  in the list stores the block_index in the split dataset.
                  For example, [[0, 1], [2]] represents the split result has
                  two datasets, datasets[0] contains block 0 and 1; and
                  datasets[1] contains block 2.
          """
          num_blocks = len(block_node_ids)
          ds = ray.data.range(num_blocks, parallelism=num_blocks)
          blocks = ds.get_internal_block_refs()
          assert len(block_node_ids) == len(blocks)
          actors = [Actor.remote() for i in range(len(actor_node_ids))]
          with patch(
              "ray.experimental.get_object_locations"
          ) as location_mock:
              with patch("ray._private.state.actors") as state_mock:
                  block_locations = {}
                  for i, node_id in enumerate(block_node_ids):
                      if node_id:
                          block_locations[blocks[i]] = {
                              "node_ids": [node_id]
                          }
                  location_mock.return_value = block_locations
  
                  actor_state = {}
                  for i, node_id in enumerate(actor_node_ids):
                      actor_state[actors[i]._actor_id.hex()] = {
                          "Address": {"NodeID": node_id}
                      }
  
                  state_mock.return_value = actor_state
  
                  datasets = ds.split(
                      len(actors), locality_hints=actors
                  )
                  assert len(datasets) == len(actors)
                  for i in range(len(actors)):
                      assert {
                          blocks[j] for j in expected_split_result[i]
                      } == set(datasets[i].get_internal_block_refs())
  
      assert_split_assignment(
          ["node2", "node1", "node1"], ["node1", "node2"], [[1, 2], [0]]
      )
      assert_split_assignment(
          ["node1", "node1", "node1"], ["node1", "node2"], [[2, 1], [0]]
      )
      assert_split_assignment(
          ["node2", "node2", None], ["node1", "node2"], [[0, 2], [1]]
      )
      assert_split_assignment(
          ["node2", "node2", None], [None, None], [[2, 1], [0]]
      )
      assert_split_assignment(
          ["n1", "n2", "n3", "n1", "n2"],
          ["n1", "n2"],
          [[0, 2, 3], [1, 4]],
      )
  
      assert_split_assignment(
          ["n1", "n2"], ["n1", "n2", "n3"], [[0], [1], []]
      )
  
      # perfect split:
      #
      # split 300 blocks
      #   with node_ids interleaving between "n0", "n1", "n2"
      #
      # to 3 actors
      #   with has node_id "n1", "n2", "n0"
      #
      # expect that block 1, 4, 7... are assigned to actor with node_id n1
      #             block 2, 5, 8... are assigned to actor with node_id n2
      #             block 0, 3, 6... are assigned to actor with node_id n0
      assert_split_assignment(
          ["n0", "n1", "n2"] * 100,
          ["n1", "n2", "n0"],
          [range(1, 300, 3), range(2, 300, 3), range(0, 300, 3)],
      )
  
      # even split regardless of locality:
      #
      # split 301 blocks
      #   with block 0 to block 50 on "n0",
      #        block 51 to block 300 on "n1"
      #
      # to 3 actors
      #   with node_ids "n1", "n2", "n0"
      #
      # expect that block 200 to block 300 are assigned to actor with node_id n1
      #             block 100 to block 199 are assigned to actor with node_id n2
      #             block 0 to block 99 are assigned to actor with node_id n0
      assert_split_assignment(
          ["n0"] * 50 + ["n1"] * 251,
          ["n1", "n2", "n0"],
          [
              range(200, 301),
              range(100, 200),
              list(range(0, 50)) + list(range(50, 100)),
          ],
      )
  
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 11 for calling a global function with example usage: test_split_hints(ray_start_regular_shared) without return types