code: def test_sort_multinode(ray_start_cluster, use_push_based_shuffle):
      cluster = ray_start_cluster
      cluster.add_node(
          resources={"bar:1": 100},
          num_cpus=10,
          _system_config={"max_direct_call_object_size": 0},
      )
      cluster.add_node(resources={"bar:2": 100}, num_cpus=10)
      cluster.add_node(resources={"bar:3": 100}, num_cpus=0)
  
      ray.init(cluster.address)
  
      parallelism = 100
      ds = (
          ray.data.range(1000, parallelism=parallelism)
          .random_shuffle()
          .sort("id")
      )
      for i, row in enumerate(ds.iter_rows()):
          assert row["id"] == i
  
imports: import random
  from collections import defaultdict
  import numpy
  import pandas
  import pyarrow
  import pytest
  import ray
  from ray.data import Dataset
  from ray.data._internal.planner.exchange.push_based_shuffle_task_scheduler import PushBasedShuffleTaskScheduler
  from ray.data._internal.sort import SortKey
  from ray.data.block import BlockAccessor
  from ray.data.context import DataContext
  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 12 for calling a global function with example usage: test_sort_multinode(ray_start_cluster, use_push_based_shuffle) without return types