class_docstr: The implementation for distributed sort tasks.
  
  The algorithm is similar to [External Merge Sort]
  (https://en.wikipedia.org/wiki/External_sorting).
  Sorting is done in 3 steps: sampling, sorting individual blocks, and
  merging sorted blocks.
  
  Sampling (`sample_boundaries`): we get a number of sample items from each block,
  sort them, and use them to compute boundaries that would partition all items into
  approximately equal ranges.
  
  Sorting (`map`): each block is sorted locally, then partitioned into smaller
  blocks according to the boundaries. Each partitioned block is passed to a merge
  task.
  
  Merging (`reduce`): a merge task would receive a block from every worker that
  consists of items in a certain range. It then merges the sorted blocks into one
  sorted block and becomes part of the new, sorted block.
code: def __init__(
      self,
      boundaries: List[T],
      sort_key: SortKey,
  ):
      super().__init__(
          map_args=[boundaries, sort_key],
          reduce_args=[sort_key],
      )
  
file_path: ray/python/ray/data/_internal/planner/exchange/sort_task_spec.py
imports: from typing import List, Tuple, TypeVar, Union
  import numpy
  from ray.data._internal.delegating_block_builder import DelegatingBlockBuilder
  from ray.data._internal.planner.exchange.interfaces import ExchangeTaskSpec
  from ray.data._internal.progress_bar import ProgressBar
  from ray.data._internal.remote_fn import cached_remote_fn
  from ray.data._internal.sort import SortKey
  from ray.data.block import Block, BlockAccessor, BlockExecStats, BlockMetadata
  from ray.types import ObjectRef
label_desc: Use image node_id 1 to create a new SortTaskSpec object from inherited base classes: ExchangeTaskSpec with example: obj = SortTaskSpec(boundaries, sort_key)