docstr: Prune unused columns from block before aggregate.
docstr_tok: ["Prune","unused","columns","from","block","before","aggregate","."]
code: def _prune_unused_columns(
      block: Block,
      key: str,
      aggs: Tuple[AggregateFn],
  ) -> Block:
      """Prune unused columns from block before aggregate."""
      prune_columns = True
      columns = set()
  
      if isinstance(key, str):
          columns.add(key)
      elif callable(key):
          prune_columns = False
  
      for agg in aggs:
          if isinstance(agg, _AggregateOnKeyBase) and isinstance(
              agg._key_fn, str
          ):
              columns.add(agg._key_fn)
          elif not isinstance(agg, Count):
              # Don't prune columns if any aggregate key is not string.
              prune_columns = False
  
      block_accessor = BlockAccessor.for_block(block)
      if (
          prune_columns
          and isinstance(block_accessor, TableBlockAccessor)
          and block_accessor.num_rows() > 0
      ):
          return block_accessor.select(list(columns))
      else:
          return block
  
imports: from typing import Any, Dict, Iterable, List, Optional, Tuple, Union
  from ._internal.table_block import TableBlockAccessor
  from ray.data._internal import sort
  from ray.data._internal.compute import ComputeStrategy
  from ray.data._internal.delegating_block_builder import DelegatingBlockBuilder
  from ray.data._internal.execution.interfaces import TaskContext
  from ray.data._internal.logical.interfaces import LogicalPlan
  from ray.data._internal.logical.operators.all_to_all_operator import Aggregate
  from ray.data._internal.plan import AllToAllStage
  from ray.data._internal.push_based_shuffle import PushBasedShufflePlan
  from ray.data._internal.shuffle import ShuffleOp, SimpleShufflePlan
  from ray.data._internal.sort import SortKey
  from ray.data.aggregate import AggregateFn, Count, Max, Mean, Min, Std, Sum
  from ray.data.aggregate._aggregate import _AggregateOnKeyBase
  from ray.data.block import Block, BlockAccessor, BlockExecStats, BlockMetadata, KeyType, UserDefinedFunction
  from ray.data.context import DataContext
  from ray.data.dataset import DataBatch, Dataset
  from ray.util.annotations import PublicAPI
label_desc: Use image node_id 3 for calling the _GroupbyOp obj's underlying member method code with example usage: obj._prune_unused_columns(block, key, aggs) and returns: block_accessor, block