code: def test_sort_simple(ray_start_regular, use_push_based_shuffle):
      num_items = 100
      parallelism = 4
      xs = list(range(num_items))
      random.shuffle(xs)
      ds = ray.data.from_items(xs, parallelism=parallelism)
      assert extract_values(
          "item", ds.sort("item").take(num_items)
      ) == list(range(num_items))
      # Make sure we have rows in each block.
      assert (
          len([n for n in ds.sort("item")._block_num_rows() if n > 0])
          == parallelism
      )
  
      assert extract_values(
          "item", ds.sort("item", descending=True).take(num_items)
      ) == list(reversed(range(num_items)))
  
      # Test empty dataset.
      ds = ray.data.from_items([])
      s1 = ds.sort("item")
      assert s1.count() == 0
      assert s1.take() == ds.take()
      ds = ray.data.range(10).filter(lambda r: r["id"] > 10).sort("id")
      assert ds.count() == 0
  
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 2 for calling a global function with example usage: test_sort_simple(ray_start_regular, use_push_based_shuffle) without return types