code: def test_sort_with_specified_boundaries(
      ray_start_regular, descending, boundaries
  ):
      num_items = 1000
      ds = ray.data.range(num_items)
      ds = ds.sort("id", descending, boundaries).materialize()
  
      items = range(num_items)
      boundaries = (
          [0] + sorted([round(b) for b in boundaries]) + [num_items]
      )
      expected_blocks = [
          items[boundaries[i] : boundaries[i + 1]]
          for i in range(len(boundaries) - 1)
      ]
      if descending:
          expected_blocks = [
              list(reversed(block))
              for block in reversed(expected_blocks)
          ]
  
      blocks = list(ds.iter_batches(batch_size=None))
      assert len(blocks) == len(expected_blocks)
      for block, expected_block in zip(blocks, expected_blocks):
          assert np.all(block["id"] == expected_block)
  
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 1 for calling a global function with example usage: test_sort_with_specified_boundaries(ray_start_regular, descending, boundaries) without return types