docstr: Read in a pyarrow table
docstr_tok: ["Read","in","a","pyarrow","table"]
code: def _read_table(
      cls,
      path_or_frag,
      fs,
      row_groups,
      columns,
      schema,
      filters,
      partitions,
      partition_keys,
      **kwargs,
  ):
      """Read in a pyarrow table"""
  
      if isinstance(path_or_frag, pa_ds.ParquetFileFragment):
          frag = path_or_frag
  
      else:
          frag = None
  
          # Check if we have partitioning information.
          # Not supported with fastparquet.
          partitioning = kwargs.get("dataset", {}).get(
              "partitioning", None
          )
  
          # Check if we need to generate a fragment.
          # NOTE: We only need a fragment if we are doing row-wise
          # filtering, or if we are missing necessary information
          # about the hive/directory partitioning. For the case
          # of filtering, we only need a fragment if we are applying
          # filters to "un-partitioned" columns. Partitioned-column
          # filters should have been applied earlier.
          missing_partitioning_info = (
              # Need to discover partition_keys
              (partitions and partition_keys is None)
              # Need to apply custom partitioning schema
              or (
                  partitioning
                  and not isinstance(partitioning, (str, list))
              )
          )
          if missing_partitioning_info or _need_filtering(
              filters, partition_keys
          ):
              # Convert the path and row-group IDs to a single fragment
              ds = pa_ds.dataset(
                  path_or_frag,
                  filesystem=_wrapped_fs(fs),
                  **_process_kwargs(**kwargs.get("dataset", {})),
              )
              frags = list(ds.get_fragments())
              assert len(frags) == 1
              frag = (
                  _frag_subset(frags[0], row_groups)
                  if row_groups != [None]
                  else frags[0]
              )
  
              # Extract hive-partition keys, and make sure they
              # are ordered the same as they are in `partitions`
              raw_keys = pa_ds._get_partition_keys(
                  frag.partition_expression
              )
              partition_keys = [
                  (hive_part.name, raw_keys[hive_part.name])
                  for hive_part in partitions
              ]
  
      if frag:
          cols = []
          for name in columns:
              if name is None:
                  if "__index_level_0__" in schema.names:
                      columns.append("__index_level_0__")
              else:
                  cols.append(name)
  
          arrow_table = frag.to_table(
              use_threads=False,
              schema=schema,
              columns=cols,
              filter=_filters_to_expression(filters)
              if filters
              else None,
          )
      else:
          arrow_table = _read_table_from_path(
              path_or_frag,
              fs,
              row_groups,
              columns,
              schema,
              filters,
              **kwargs,
          )
  
      # For pyarrow.dataset api, if we did not read directly from
      # fragments, we need to add the partitioned columns here.
      if partitions and isinstance(partitions, list):
          keys_dict = {k: v for (k, v) in partition_keys}
          for partition in partitions:
              if partition.name not in arrow_table.schema.names:
                  # We read from file paths, so the partition
                  # columns may NOT be in our table yet.
                  cat = keys_dict.get(partition.name, None)
                  if not len(partition.keys):
                      arr = pa.array(np.full(len(arrow_table), cat))
                  else:
                      cat_ind = np.full(
                          len(arrow_table),
                          partition.keys.get_loc(cat),
                          dtype="i4",
                      )
                      arr = pa.DictionaryArray.from_arrays(
                          cat_ind, pa.array(partition.keys)
                      )
                  arrow_table = arrow_table.append_column(
                      partition.name, arr
                  )
  
      return arrow_table
  
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 14 for calling the ArrowDatasetEngine obj's underlying member method code with example usage: obj._read_table(cls, path_or_frag, fs, row_groups, columns, schema, filters, partitions, partition_keys) and returns: arrow_table