docstr: pyarrow.dataset version of _collect_dataset_info
  Use pyarrow.dataset API to construct a dictionary of all
  general information needed to read the dataset.
docstr_tok: ["pyarrow.dataset","version","of","_collect_dataset_info","Use","pyarrow.dataset","API","to","construct","a","dictionary","of","all","general","information","needed","to","read","the","dataset","."]
code: def _collect_dataset_info(
      cls,
      paths,
      fs,
      categories,
      index,
      gather_statistics,
      filters,
      split_row_groups,
      blocksize,
      aggregate_files,
      ignore_metadata_file,
      metadata_task_size,
      parquet_file_extension,
      kwargs,
  ):
      """pyarrow.dataset version of _collect_dataset_info
      Use pyarrow.dataset API to construct a dictionary of all
      general information needed to read the dataset.
      """
  
      # Use pyarrow.dataset API
      ds = None
      valid_paths = (
          None  # Only used if `paths` is a list containing _metadata
      )
  
      # Extract dataset-specific options
      _dataset_kwargs = kwargs.pop("dataset", {})
  
      if "partitioning" not in _dataset_kwargs:
          _dataset_kwargs["partitioning"] = "hive"
  
      if "format" not in _dataset_kwargs:
          _dataset_kwargs["format"] = pa_ds.ParquetFileFormat()
      _processed_dataset_kwargs = _process_kwargs(**_dataset_kwargs)
  
      # Case-dependent pyarrow.dataset creation
      has_metadata_file = False
      if len(paths) == 1 and fs.isdir(paths[0]):
          # Use _analyze_paths to avoid relative-path
          # problems (see GH#5608)
          paths, base, fns = _sort_and_analyze_paths(paths, fs)
          paths = fs.sep.join([base, fns[0]])
  
          meta_path = fs.sep.join([paths, "_metadata"])
          if not ignore_metadata_file and fs.exists(meta_path):
              # Use _metadata file
              ds = pa_ds.parquet_dataset(
                  meta_path,
                  filesystem=_wrapped_fs(fs),
                  **_processed_dataset_kwargs,
              )
              has_metadata_file = True
          elif parquet_file_extension:
              # Need to materialize all paths if we are missing the _metadata file
              # Raise error if all files have been filtered by extension
              len0 = len(paths)
              paths = [
                  path
                  for path in fs.find(paths)
                  if path.endswith(parquet_file_extension)
              ]
              if len0 and paths == []:
                  raise ValueError(
                      "No files satisfy the `parquet_file_extension` criteria "
                      f"(files must end with {parquet_file_extension})."
                  )
  
      elif len(paths) > 1:
          paths, base, fns = _sort_and_analyze_paths(paths, fs)
          meta_path = fs.sep.join([base, "_metadata"])
          if "_metadata" in fns:
              # Pyarrow cannot handle "_metadata" when `paths` is a list
              # Use _metadata file
              if not ignore_metadata_file:
                  ds = pa_ds.parquet_dataset(
                      meta_path,
                      filesystem=_wrapped_fs(fs),
                      **_processed_dataset_kwargs,
                  )
                  has_metadata_file = True
  
              # Populate valid_paths, since the original path list
              # must be used to filter the _metadata-based dataset
              fns.remove("_metadata")
              valid_paths = fns
  
      # Final "catch-all" pyarrow.dataset call
      if ds is None:
          ds = pa_ds.dataset(
              paths,
              filesystem=_wrapped_fs(fs),
              **_processed_dataset_kwargs,
          )
  
      # Get file_frag sample and extract physical_schema
      try:
          file_frag = next(ds.get_fragments())
          physical_schema = file_frag.physical_schema
      except StopIteration:
          file_frag = None
          physical_schema = ds.schema
  
      # Set split_row_groups for desired partitioning behavior
      #
      # Expected behavior for split_row_groups + blocksize combinations:
      # +======+==================+===========+=============================+
      # | Case | split_row_groups | blocksize | Behavior                    |
      # +======+==================+===========+=============================+
      # |  A   |  "infer"         |  not None | Go to E or G (using md)     |
      # +------+------------------+-----------+-----------------------------+
      # |  B   |  "infer"         |  None     | Go to H                     |
      # +------+------------------+-----------+-----------------------------+
      # |  C   |  "adaptive"      |  not None | Go to E                     |
      # +------+------------------+-----------+-----------------------------+
      # |  D   |  "adaptive"      |  None     | Go to H                     |
      # +======+==================+===========+=============================+
      # |  E*  |  True            |  not None | Adaptive partitioning       |
      # +------+------------------+-----------+-----------------------------+
      # |  F   |  True            |  None     | 1 row-group per partition   |
      # +------+------------------+-----------+-----------------------------+
      # |  G*  |  False           |  not None | 1+ full files per partition |
      # +------+------------------+-----------+-----------------------------+
      # |  H   |  False           |  None     | 1 full file per partition   |
      # +------+------------------+-----------+-----------------------------+
      # |  I   |  n               |  N/A      | n row-groups per partition  |
      # +======+==================+===========+=============================+
      # NOTES:
      # - Adaptive partitioning (E) means that the individual size of each
      #   row-group will be accounted for when deciding how many row-groups
      #   to map to each output partition.
      # - E, G and I will only aggregate data from multiple files into the
      #   same output partition if `bool(aggregate_files) == True`.
      # - Default partitioning will correspond to either E or G. All other
      #   behavior requires user input.
  
      if split_row_groups == "infer":
          if blocksize:
              # Sample row-group sizes in first file
              if file_frag is None:
                  # Empty dataset
                  split_row_groups = False
              else:
                  split_row_groups = _infer_split_row_groups(
                      [
                          rg.total_byte_size
                          for rg in file_frag.row_groups
                      ],
                      blocksize,
                      bool(aggregate_files),
                  )
          else:
              split_row_groups = False
  
      if split_row_groups == "adaptive":
          if blocksize:
              split_row_groups = True
          else:
              split_row_groups = False
  
      # Note on (hive) partitioning information:
      #
      #    - "partition_obj" : (list of PartitionObj) This is a list of
      #          simple objects providing `name` and `keys` attributes
      #          for each partition column.
      #    - "partition_names" : (list)  This is a list containing the
      #          names of partitioned columns.
      #
      partition_obj, partition_names = [], []
      if ds.partitioning and ds.partitioning.schema:
          partition_names = list(ds.partitioning.schema.names)
          for i, name in enumerate(partition_names):
              dictionary = (
                  ds.partitioning.dictionaries[i]
                  if ds.partitioning.dictionaries
                  else None
              )
              partition_obj.append(
                  PartitionObj(
                      name,
                      (
                          pd.Series([], dtype="object")
                          if dictionary is None
                          else dictionary.to_pandas()
                      ),
                  )
              )
  
      # Check the `aggregate_files` setting
      aggregation_depth = _get_aggregation_depth(
          aggregate_files, partition_names
      )
  
      return {
          "ds": ds,
          "physical_schema": physical_schema,
          "has_metadata_file": has_metadata_file,
          "schema": ds.schema,
          "fs": fs,
          "valid_paths": valid_paths,
          "gather_statistics": gather_statistics,
          "categories": categories,
          "index": index,
          "filters": filters,
          "split_row_groups": split_row_groups,
          "blocksize": blocksize,
          "aggregate_files": aggregate_files,
          "aggregation_depth": aggregation_depth,
          "partitions": partition_obj,
          "partition_names": partition_names,
          "metadata_task_size": metadata_task_size,
          "kwargs": {
              "dataset": _dataset_kwargs,
              "convert_string": pyarrow_strings_enabled(),
              **kwargs,
          },
      }
  
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 9 for calling the ArrowDatasetEngine obj's underlying member method code with example usage: obj._collect_dataset_info(cls, paths, fs, categories, index, gather_statistics, filters, split_row_groups, blocksize, aggregate_files, ignore_metadata_file, metadata_task_size, parquet_file_extension, kwargs) and returns: dict