code: def _collect_file_parts(
      cls,
      files_or_frags,
      dataset_info_kwargs,
  ):
      # Collect necessary information from dataset_info
      fs = dataset_info_kwargs["fs"]
      split_row_groups = dataset_info_kwargs["split_row_groups"]
      gather_statistics = dataset_info_kwargs["gather_statistics"]
      partitions = dataset_info_kwargs["partitions"]
      dataset_options = dataset_info_kwargs["dataset_options"]
  
      # Make sure we are processing a non-empty list
      if not isinstance(files_or_frags, list):
          files_or_frags = [files_or_frags]
      elif not files_or_frags:
          return [], []
  
      # Make sure we are starting with file fragments
      if isinstance(files_or_frags[0], str):
          # Check if we are using a simple file-partition map
          # without requiring any file or row-group statistics
          if (
              not (split_row_groups or partitions)
              and gather_statistics is False
          ):
              # Cool - We can return immediately
              return [
                  {"piece": (file_or_frag, None, None)}
                  for file_or_frag in files_or_frags
              ], None
  
          # Need more information - convert the path to a fragment
          file_frags = list(
              pa_ds.dataset(
                  files_or_frags,
                  filesystem=_wrapped_fs(fs),
                  **_process_kwargs(**dataset_options),
              ).get_fragments()
          )
      else:
          file_frags = files_or_frags
  
      # Collect settings from dataset_info
      filters = dataset_info_kwargs["filters"]
      ds_filters = dataset_info_kwargs["ds_filters"]
      schema = dataset_info_kwargs["schema"]
      stat_col_indices = dataset_info_kwargs["stat_col_indices"]
      aggregation_depth = dataset_info_kwargs["aggregation_depth"]
      blocksize = dataset_info_kwargs["blocksize"]
  
      # Intialize row-group and statistics data structures
      file_row_groups = defaultdict(list)
      file_row_group_stats = defaultdict(list)
      file_row_group_column_stats = defaultdict(list)
      single_rg_parts = int(split_row_groups) == 1
      hive_partition_keys = {}
      cmax_last = {}
      for file_frag in file_frags:
          fpath = file_frag.path
  
          # Extract hive-partition keys, and make sure they
          # are orederd the same as they are in `partitions`
          raw_keys = pa_ds._get_partition_keys(
              file_frag.partition_expression
          )
          hive_partition_keys[fpath] = [
              (hive_part.name, raw_keys[hive_part.name])
              for hive_part in partitions
          ]
  
          for frag in file_frag.split_by_row_group(
              ds_filters, schema=schema
          ):
              row_group_info = frag.row_groups
              if gather_statistics or split_row_groups:
                  # If we are gathering statistics or splitting by
                  # row-group, we may need to ensure our fragment
                  # metadata is complete.
                  if row_group_info is None:
                      frag.ensure_complete_metadata()
                      row_group_info = frag.row_groups
                  if not len(row_group_info):
                      continue
              else:
                  file_row_groups[fpath] = [None]
                  continue
              for row_group in row_group_info:
                  file_row_groups[fpath].append(row_group.id)
                  if gather_statistics:
                      statistics = _get_rg_statistics(
                          row_group, list(stat_col_indices)
                      )
                      if single_rg_parts:
                          s = {
                              "file_path_0": fpath,
                              "num-rows": row_group.num_rows,
                              "total_byte_size": row_group.total_byte_size,
                              "columns": [],
                          }
                      else:
                          s = {
                              "num-rows": row_group.num_rows,
                              "total_byte_size": row_group.total_byte_size,
                          }
                      cstats = []
                      for name in stat_col_indices.keys():
                          if name in statistics:
                              cmin = statistics[name]["min"]
                              cmax = statistics[name]["max"]
                              null_count = statistics[name][
                                  "null_count"
                              ]
                              cmin = (
                                  pd.Timestamp(cmin)
                                  if isinstance(cmin, datetime)
                                  else cmin
                              )
                              cmax = (
                                  pd.Timestamp(cmax)
                                  if isinstance(cmax, datetime)
                                  else cmax
                              )
                              last = cmax_last.get(name, None)
                              if not (
                                  filters
                                  or (
                                      blocksize
                                      and split_row_groups is True
                                  )
                                  or aggregation_depth
                              ):
                                  # Only think about bailing if we don't need
                                  # stats for filtering
                                  if cmin is None or (
                                      last and cmin < last
                                  ):
                                      # We are collecting statistics for divisions
                                      # only (no filters) - Column isn't sorted, or
                                      # we have an all-null partition, so lets bail.
                                      #
                                      # Note: This assumes ascending order.
                                      #
                                      gather_statistics = False
                                      file_row_group_stats = {}
                                      file_row_group_column_stats = {}
                                      break
  
                              if single_rg_parts:
                                  s["columns"].append(
                                      {
                                          "name": name,
                                          "min": cmin,
                                          "max": cmax,
                                          "null_count": null_count,
                                      }
                                  )
                              else:
                                  cstats += [cmin, cmax, null_count]
                              cmax_last[name] = cmax
                          else:
                              if single_rg_parts:
                                  s["columns"].append({"name": name})
                              else:
                                  cstats += [None, None, None]
                      if gather_statistics:
                          file_row_group_stats[fpath].append(s)
                          if not single_rg_parts:
                              file_row_group_column_stats[fpath].append(
                                  tuple(cstats)
                              )
  
      # Check if we have empty parts to return
      if not file_row_groups:
          return [], []
  
      # Convert organized row-groups to parts
      return _row_groups_to_parts(
          gather_statistics,
          split_row_groups,
          aggregation_depth,
          file_row_groups,
          file_row_group_stats,
          file_row_group_column_stats,
          stat_col_indices,
          cls._make_part,
          make_part_kwargs={
              "fs": fs,
              "partition_keys": hive_partition_keys,
              "partition_obj": partitions,
              "data_path": "",
          },
      )
  
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 12 for calling the ArrowDatasetEngine obj's underlying member method code with example usage: obj._collect_file_parts(cls, files_or_frags, dataset_info_kwargs) and returns: _row_groups_to_parts, list,  list, list,  list, unknown,  None