docstr: Read in a single output partition
docstr_tok: ["Read","in","a","single","output","partition"]
code: def read_partition(
      cls,
      fs,
      pieces,
      columns,
      index,
      dtype_backend=None,
      categories=(),
      partitions=(),
      filters=None,
      schema=None,
      **kwargs,
  ):
      """Read in a single output partition"""
  
      if isinstance(index, list):
          for level in index:
              # unclear if we can use set ops here. I think the order matters.
              # Need the membership test to avoid duplicating index when
              # we slice with `columns` later on.
              if level not in columns:
                  columns.append(level)
  
      # Ensure `columns` and `partitions` do not overlap
      columns_and_parts = columns.copy()
      if not isinstance(partitions, (list, tuple)):
          if columns_and_parts and partitions:
              for part_name in partitions.partition_names:
                  if part_name in columns:
                      columns.remove(part_name)
                  else:
                      columns_and_parts.append(part_name)
              columns = columns or None
  
      # Always convert pieces to list
      if not isinstance(pieces, list):
          pieces = [pieces]
  
      tables = []
      multi_read = len(pieces) > 1
      for piece in pieces:
          if isinstance(piece, str):
              # `piece` is a file-path string
              path_or_frag = piece
              row_group = None
              partition_keys = None
          else:
              # `piece` contains (path, row_group, partition_keys)
              (path_or_frag, row_group, partition_keys) = piece
  
          # Convert row_group to a list and be sure to
          # check if msgpack converted it to a tuple
          if isinstance(row_group, tuple):
              row_group = list(row_group)
          if not isinstance(row_group, list):
              row_group = [row_group]
  
          # Read in arrow table and convert to pandas
          arrow_table = cls._read_table(
              path_or_frag,
              fs,
              row_group,
              columns,
              schema,
              filters,
              partitions,
              partition_keys,
              **kwargs,
          )
          if multi_read:
              tables.append(arrow_table)
  
      if multi_read:
          arrow_table = pa.concat_tables(tables)
  
      # Convert to pandas
      df = cls._arrow_table_to_pandas(
          arrow_table, categories, dtype_backend=dtype_backend, **kwargs
      )
  
      # For pyarrow.dataset api, need to convert partition columns
      # to categorigal manually for integer types.
      if partitions and isinstance(partitions, list):
          for partition in partitions:
              if (
                  len(partition.keys)
                  and df[partition.name].dtype.name != "category"
              ):
                  # We read directly from fragments, so the partition
                  # columns are already in our dataframe.  We just
                  # need to convert non-categorical types.
                  df[partition.name] = pd.Series(
                      pd.Categorical(
                          categories=partition.keys,
                          values=df[partition.name].values,
                      ),
                      index=df.index,
                  )
  
      # Note that `to_pandas(ignore_metadata=False)` means
      # pyarrow will use the pandas metadata to set the index.
      index_in_columns_and_parts = set(df.index.names).issubset(
          set(columns_and_parts)
      )
      if not index:
          if index_in_columns_and_parts:
              # User does not want to set index and a desired
              # column/partition has been set to the index
              df.reset_index(drop=False, inplace=True)
          else:
              # User does not want to set index and an
              # "unwanted" column has been set to the index
              df.reset_index(drop=True, inplace=True)
      else:
          if (
              set(df.index.names) != set(index)
              and index_in_columns_and_parts
          ):
              # The wrong index has been set and it contains
              # one or more desired columns/partitions
              df.reset_index(drop=False, inplace=True)
          elif index_in_columns_and_parts:
              # The correct index has already been set
              index = False
              columns_and_parts = list(
                  set(columns_and_parts) - set(df.index.names)
              )
      df = df[list(columns_and_parts)]
  
      if index:
          df = df.set_index(index)
      return df
  
imports: from __future__ import annotations
  import json
  import operator
  import textwrap
  import warnings
  from collections import defaultdict
  from datetime import datetime
  from functools import reduce
  import numpy
  import pandas
  import pyarrow
  import pyarrow.parquet
  from fsspec.core import expand_paths_if_needed, stringify_path
  from fsspec.implementations.arrow import ArrowFSWrapper
  from pyarrow import dataset
  from pyarrow import fs
  import dask
  from dask.base import tokenize
  from dask.core import flatten
  from dask.dataframe.backends import pyarrow_schema_dispatch
  from dask.dataframe.io.parquet.utils import Engine, _get_aggregation_depth, _infer_split_row_groups, _normalize_index_columns, _process_open_file_options, _row_groups_to_parts, _set_gather_statistics, _set_metadata_task_size, _sort_and_analyze_paths
  from dask.dataframe.io.utils import _get_pyarrow_dtypes, _is_local_fs, _open_input_files
  from dask.dataframe.utils import clear_known_categories, pyarrow_strings_enabled
  from dask.delayed import Delayed
  from dask.utils import getargspec, natural_sort_key
label_desc: Use image node_id 4 for calling the ArrowDatasetEngine obj's underlying member method code with example usage: obj.read_partition(cls, fs, pieces, columns, index, dtype_backend, categories, partitions, filters, schema) and returns: df