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 map(
      idx: int,
      block: Block,
      output_num_blocks: int,
      boundaries: List[T],
      sort_key: SortKey,
  ) -> List[Union[BlockMetadata, Block]]:
      stats = BlockExecStats.builder()
      out = BlockAccessor.for_block(block).sort_and_partition(
          boundaries, sort_key
      )
      meta = BlockAccessor.for_block(block).get_metadata(
          input_files=None, exec_stats=stats.build()
      )
      return out + [meta]
  
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 2 for calling the SortTaskSpec obj's underlying member method code with example usage: obj.map(idx, block, output_num_blocks, boundaries, sort_key) and returns: unknown