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,
  ):
      # Define the parquet-file (pf) object to use for metadata,
      # Also, initialize `parts`.  If `parts` is populated here,
      # then each part will correspond to a file.  Otherwise, each part will
      # correspond to a row group (populated later).
  
      # Extract dataset-specific options
      dataset_kwargs = kwargs.pop("dataset", {})
  
      parts = []
      _metadata_exists = False
      if len(paths) == 1 and fs.isdir(paths[0]):
          # This is a directory.
          # Check if _metadata and/or _common_metadata files exists
          base = paths[0]
          _metadata_exists = True
          if not ignore_metadata_file:
              _metadata_exists = fs.isfile(
                  fs.sep.join([base, "_metadata"])
              )
  
          # Find all files if we are not using a _metadata file
          if ignore_metadata_file or not _metadata_exists:
              # For now, we need to discover every file under paths[0]
              paths, base, fns = _sort_and_analyze_paths(
                  fs.find(base), fs, root=base
              )
              _update_paths = False
              for fn in ["_metadata", "_common_metadata"]:
                  try:
                      fns.remove(fn)
                      _update_paths = True
                  except ValueError:
                      pass
              if _update_paths:
                  paths = [fs.sep.join([base, fn]) for fn in fns]
              _metadata_exists = False
          if _metadata_exists:
              # Using _metadata file (best-case scenario)
              pf = ParquetFile(
                  fs.sep.join([base, "_metadata"]),
                  open_with=fs.open,
                  **dataset_kwargs,
              )
          else:
              # Use 0th file
              # Note that "_common_metadata" can cause issues for
              # partitioned datasets.
              if parquet_file_extension:
                  # Raise error if all files have been filtered by extension
                  len0 = len(paths)
                  paths = [
                      path
                      for path in paths
                      if path.endswith(parquet_file_extension)
                  ]
                  fns = [
                      fn
                      for fn in fns
                      if fn.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})."
                      )
              pf = ParquetFile(
                  paths[:1],
                  open_with=fs.open,
                  root=base,
                  **dataset_kwargs,
              )
              scheme = get_file_scheme(fns)
              pf.file_scheme = scheme
              pf.cats = paths_to_cats(fns, scheme)
              if not gather_statistics:
                  parts = [fs.sep.join([base, fn]) for fn in fns]
      else:
          # This is a list of files
          paths, base, fns = _sort_and_analyze_paths(paths, fs)
  
          # Check if _metadata is in paths, and
          # remove it if ignore_metadata_file=True
          _metadata_exists = "_metadata" in fns
          if _metadata_exists and ignore_metadata_file:
              fns.remove("_metadata")
              _metadata_exists = False
          paths = [fs.sep.join([base, fn]) for fn in fns]
  
          if _metadata_exists:
              # We have a _metadata file, lets use it
              pf = ParquetFile(
                  fs.sep.join([base, "_metadata"]),
                  open_with=fs.open,
                  **dataset_kwargs,
              )
          else:
              # Rely on metadata for 0th file.
              # Will need to pass a list of paths to read_partition
              scheme = get_file_scheme(fns)
              pf = ParquetFile(
                  paths[:1],
                  open_with=fs.open,
                  root=base,
                  **dataset_kwargs,
              )
              pf.file_scheme = scheme
              pf.cats = paths_to_cats(fns, scheme)
              if not gather_statistics:
                  parts = paths.copy()
  
      # Check the `aggregate_files` setting
      aggregation_depth = _get_aggregation_depth(
          aggregate_files,
          list(pf.cats),
      )
  
      # Ensure that there is no overlap between partition columns
      # and explicit columns in `pf`
      if pf.cats:
          _partitions = [p for p in pf.cats if p not in pf.columns]
          if not _partitions:
              pf.cats = {}
          elif len(_partitions) != len(pf.cats):
              raise ValueError(
                  "No partition-columns should be written in the \n"
                  "file unless they are ALL written in the file.\n"
                  "This restriction is removed as of fastparquet 0.8.4\n"
                  "columns: {} | partitions: {}".format(
                      pf.columns, pf.cats.keys()
                  )
              )
  
      # 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
              pf_sample = ParquetFile(
                  paths[0],
                  open_with=fs.open,
                  **dataset_kwargs,
              )
              split_row_groups = _infer_split_row_groups(
                  [rg.total_byte_size for rg in pf_sample.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
  
      return {
          "pf": pf,
          "paths": paths,
          "has_metadata_file": _metadata_exists,
          "parts": parts,
          "base": base,
          "fs": fs,
          "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,
          "metadata_task_size": metadata_task_size,
          "kwargs": {
              "dataset": dataset_kwargs,
              **kwargs,
          },
      }
  
imports: from __future__ import annotations
  import copy
  import pickle
  import threading
  import warnings
  from collections import OrderedDict, defaultdict
  from contextlib import ExitStack
  import numpy
  import pandas
  import tlz
  from packaging.version import parse
  from dask.core import flatten
  from dask.dataframe._compat import PANDAS_GE_201
  from dask.base import tokenize
  from dask.dataframe.io.parquet.utils import Engine, _get_aggregation_depth, _infer_split_row_groups, _normalize_index_columns, _parse_pandas_metadata, _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 _is_local_fs, _meta_from_dtypes, _open_input_files
  from dask.dataframe.utils import UNKNOWN_CATEGORIES
  from dask.delayed import Delayed
  from dask.utils import natural_sort_key
label_desc: Use image node_id 4 for calling the FastParquetEngine 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