code: def sample_boundaries(
      blocks: List[ObjectRef[Block]],
      sort_key: SortKey,
      num_reducers: int,
      ctx: Optional[TaskContext] = None,
  ) -> List[T]:
      """
      Return (num_reducers - 1) items in ascending order from the blocks that
      partition the domain into ranges with approximately equally many elements.
      Each boundary item is a tuple of a form (col1_value, col2_value, ...).
      """
      columns = sort_key.get_columns()
  
      n_samples = int(num_reducers * 10 / len(blocks))
  
      sample_block = cached_remote_fn(_sample_block)
  
      sample_results = [
          sample_block.remote(block, n_samples, sort_key)
          for block in blocks
      ]
  
      should_close_bar = True
      if ctx is not None and ctx.sub_progress_bar_dict is not None:
          bar_name = "SortSample"
          assert (
              bar_name in ctx.sub_progress_bar_dict
          ), ctx.sub_progress_bar_dict
          sample_bar = ctx.sub_progress_bar_dict[bar_name]
          should_close_bar = False
      else:
          sample_bar = ProgressBar("Sort Sample", len(sample_results))
  
      samples = sample_bar.fetch_until_complete(sample_results)
  
      if should_close_bar:
          sample_bar.close()
      del sample_results
      samples = [s for s in samples if len(s) > 0]
      # The dataset is empty
      if len(samples) == 0:
          return [None] * (num_reducers - 1)
      builder = DelegatingBlockBuilder()
      for sample in samples:
          builder.add_block(sample)
      samples = builder.build()
      sample_dict = BlockAccessor.for_block(samples).to_numpy(
          columns=columns
      )
      # Compute sorted indices of the samples. In np.lexsort last key is the
      # primary key hence have to reverse the order.
      indices = np.lexsort(list(reversed(list(sample_dict.values()))))
      # Sort each column by indices, and calculate q-ths quantile items.
      # Ignore the 1st item as it's not required for the boundary
      for k, v in sample_dict.items():
          sorted_v = v[indices]
          sample_dict[k] = [
              np.quantile(sorted_v, q, interpolation="nearest")
              for q in np.linspace(0, 1, num_reducers)
          ][1:]
      # Return the list of boundaries as tuples
      # of a form (col1_value, col2_value, ...)
      return [
          tuple(sample_dict[k][i] for k in sample_dict)
          for i in range(num_reducers - 1)
      ]
  
imports: from typing import TYPE_CHECKING, List, Optional, Tuple, TypeVar, Union
  import numpy
  from ray.data._internal.block_list import BlockList
  from ray.data._internal.delegating_block_builder import DelegatingBlockBuilder
  from ray.data._internal.execution.interfaces import TaskContext
  from ray.data._internal.progress_bar import ProgressBar
  from ray.data._internal.push_based_shuffle import PushBasedShufflePlan
  from ray.data._internal.remote_fn import cached_remote_fn
  from ray.data._internal.shuffle import ShuffleOp, SimpleShufflePlan
  from ray.data.block import Block, BlockAccessor, BlockExecStats, BlockMetadata
  from ray.data.context import DataContext
  from ray.types import ObjectRef
label_desc: Use image node_id 1 for calling a global function with example usage: sample_boundaries(blocks, sort_key, num_reducers, ctx) and returns: unknown, unknown