code: def _construct_collection_plan(cls, dataset_info):
      # Collect necessary information from dataset_info
      fs = dataset_info["fs"]
      parts = dataset_info["parts"]
      paths = dataset_info["paths"]
      filters = dataset_info["filters"]
      pf = dataset_info["pf"]
      split_row_groups = dataset_info["split_row_groups"]
      blocksize = dataset_info["blocksize"]
      gather_statistics = dataset_info["gather_statistics"]
      base_path = dataset_info["base"]
      aggregation_depth = dataset_info["aggregation_depth"]
      index_cols = dataset_info["index_cols"]
      categories = dataset_info["categories"]
      dtypes = dataset_info["dtypes"]
      categories_dict = dataset_info["categories_dict"]
      has_metadata_file = dataset_info["has_metadata_file"]
      metadata_task_size = dataset_info["metadata_task_size"]
      kwargs = dataset_info["kwargs"]
  
      # Ensure metadata_task_size is set
      # (Using config file or defaults)
      metadata_task_size = _set_metadata_task_size(
          dataset_info["metadata_task_size"], fs
      )
  
      # Determine which columns need statistics.
      # At this point, gather_statistics is only True if
      # the user specified calculate_divisions=True
      filter_columns = {
          t[0] for t in flatten(filters or [], container=list)
      }
      stat_col_indices = {}
      _index_cols = (
          index_cols
          if (gather_statistics and len(index_cols) == 1)
          else []
      )
      for i, name in enumerate(pf.columns):
          if name in _index_cols or name in filter_columns:
              stat_col_indices[name] = i
  
      # Decide final `gather_statistics` setting.
      # NOTE: The "fastparquet" engine requires statistics for
      # filtering even if the filter is on a paritioned column
      gather_statistics = _set_gather_statistics(
          gather_statistics,
          blocksize,
          split_row_groups,
          aggregation_depth,
          filter_columns,
          set(stat_col_indices) | filter_columns,
      )
  
      # Define common_kwargs
      common_kwargs = {
          "categories": categories_dict or categories,
          "root_cats": pf.cats,
          "root_file_scheme": pf.file_scheme,
          "base_path": base_path,
          **kwargs,
      }
  
      # Check if this is a very simple case where we can just
      # return the path names. This requires that `parts`
      # already be a list of paths. Also, we cannot be splitting
      # by row-group or collecting statistics.
      if (
          gather_statistics is False
          and not split_row_groups
          and isinstance(parts, list)
          and len(parts)
          and isinstance(parts[0], str)
      ):
          return (
              [{"piece": (full_path, None)} for full_path in parts],
              [],
              common_kwargs,
          )
  
      dataset_info_kwargs = {
          "fs": fs,
          "split_row_groups": split_row_groups,
          "gather_statistics": gather_statistics,
          "filters": filters,
          "dtypes": dtypes,
          "stat_col_indices": stat_col_indices,
          "aggregation_depth": aggregation_depth,
          "blocksize": blocksize,
          "root_cats": pf.cats,
          "root_file_scheme": pf.file_scheme,
          "base_path": "" if base_path is None else base_path,
          "has_metadata_file": has_metadata_file,
      }
  
      if (
          has_metadata_file
          or metadata_task_size == 0
          or metadata_task_size > len(paths)
      ):
          # Construct the output-partitioning plan on the
          # client process (in serial).  This means we have
          # a global _metadata file, or that `metadata_task_size`
          # is zero or larger than the number of files.
          pf_or_paths = pf if has_metadata_file else paths
          parts, stats = cls._collect_file_parts(
              pf_or_paths, dataset_info_kwargs
          )
  
      else:
          # We DON'T have a global _metadata file to work with.
          # We should loop over files in parallel
          parts, stats = [], []
          if paths:
              # Build and compute a task graph to construct stats/parts
              gather_parts_dsk = {}
              name = "gather-pq-parts-" + tokenize(
                  paths, dataset_info_kwargs
              )
              finalize_list = []
              for task_i, file_i in enumerate(
                  range(0, len(paths), metadata_task_size)
              ):
                  finalize_list.append((name, task_i))
                  gather_parts_dsk[finalize_list[-1]] = (
                      cls._collect_file_parts,
                      paths[file_i : file_i + metadata_task_size],
                      dataset_info_kwargs,
                  )
  
              def _combine_parts(parts_and_stats):
                  parts, stats = [], []
                  for part, stat in parts_and_stats:
                      parts += part
                      if stat:
                          stats += stat
                  return parts, stats
  
              gather_parts_dsk["final-" + name] = (
                  _combine_parts,
                  finalize_list,
              )
              parts, stats = Delayed(
                  "final-" + name, gather_parts_dsk
              ).compute()
  
      return parts, stats, common_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 6 for calling the FastParquetEngine obj's underlying member method code with example usage: obj._construct_collection_plan(cls, dataset_info) and returns: parts,  stats,  common_kwargs, unknown,  list,  common_kwargs, parts,  stats