docstr: pyarrow.dataset version of _construct_collection_plan
  Use dataset_info to construct the general plan for
  generating the output DataFrame collection.
  
  The "plan" is essentially a list (called `parts`) of
  information that is needed to produce each output partition.
  After this function is returned, the information in each
  element of `parts` will be used to produce a single Dask-
  DataFrame partition (unless some elements of `parts`
  are aggregated together in a follow-up step).
  
  This method also returns ``stats`` (which is a list of
  parquet-metadata statistics for each element of parts),
  and ``common_metadata`` (which is a dictionary of kwargs
  that should be passed to the ``read_partition`` call for
  every output partition).
docstr_tok: ["pyarrow.dataset","version","of","_construct_collection_plan","Use","dataset_info","to","construct","the","general","plan","for","generating","the","output","DataFrame","collection",".","The","``","plan","''","is","essentially","a","list","(","called","`","parts","`",")","of","information","that","is","needed","to","produce","each","output","partition",".","After","this","function","is","returned",",","the","information","in","each","element","of","`","parts","`","will","be","used","to","produce","a","single","Dask-","DataFrame","partition","(","unless","some","elements","of","`","parts","`","are","aggregated","together","in","a","follow-up","step",")",".","This","method","also","returns","``","stats","``","(","which","is","a","list","of","parquet-metadata","statistics","for","each","element","of","parts",")",",","and","``","common_metadata","``","(","which","is","a","dictionary","of","kwargs","that","should","be","passed","to","the","``","read_partition","``","call","for","every","output","partition",")","."]
code: def _construct_collection_plan(cls, dataset_info):
      """pyarrow.dataset version of _construct_collection_plan
      Use dataset_info to construct the general plan for
      generating the output DataFrame collection.
  
      The "plan" is essentially a list (called `parts`) of
      information that is needed to produce each output partition.
      After this function is returned, the information in each
      element of `parts` will be used to produce a single Dask-
      DataFrame partition (unless some elements of `parts`
      are aggregated together in a follow-up step).
  
      This method also returns ``stats`` (which is a list of
      parquet-metadata statistics for each element of parts),
      and ``common_metadata`` (which is a dictionary of kwargs
      that should be passed to the ``read_partition`` call for
      every output partition).
      """
  
      # Collect necessary dataset information from dataset_info
      ds = dataset_info["ds"]
      fs = dataset_info["fs"]
      filters = dataset_info["filters"]
      split_row_groups = dataset_info["split_row_groups"]
      gather_statistics = dataset_info["gather_statistics"]
      blocksize = dataset_info["blocksize"]
      aggregation_depth = dataset_info["aggregation_depth"]
      index_cols = dataset_info["index_cols"]
      schema = dataset_info["schema"]
      partition_names = dataset_info["partition_names"]
      partitions = dataset_info["partitions"]
      categories = dataset_info["categories"]
      has_metadata_file = dataset_info["has_metadata_file"]
      valid_paths = dataset_info["valid_paths"]
      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
      )
  
      # Make sure that any `in`-predicate filters have iterable values
      filter_columns = set()
      if filters is not None:
          for filter in flatten(filters, container=list):
              col, op, val = filter
              if op == "in" and not isinstance(val, (set, list, tuple)):
                  raise TypeError(
                      "Value of 'in' filter must be a list, set or tuple."
                  )
              filter_columns.add(col)
  
      # Determine which columns need statistics.
      # At this point, gather_statistics is only True if
      # the user specified calculate_divisions=True
      stat_col_indices = {}
      _index_cols = (
          index_cols
          if (gather_statistics and len(index_cols) == 1)
          else []
      )
      for i, name in enumerate(schema.names):
          if name in _index_cols or name in filter_columns:
              if name in partition_names:
                  # Partition columns wont have statistics
                  continue
              stat_col_indices[name] = i
  
      # Decide final `gather_statistics` setting
      gather_statistics = _set_gather_statistics(
          gather_statistics,
          blocksize,
          split_row_groups,
          aggregation_depth,
          filter_columns,
          set(stat_col_indices),
      )
  
      # Add common kwargs
      common_kwargs = {
          "partitions": partitions,
          "categories": categories,
          "filters": filters,
          "schema": schema,
          **kwargs,
      }
  
      # Check if this is a very simple case where we can just return
      # the path names
      if gather_statistics is False and not (
          split_row_groups or filters
      ):
          return (
              [
                  {"piece": (full_path, None, None)}
                  for full_path in sorted(
                      ds.files, key=natural_sort_key
                  )
              ],
              [],
              common_kwargs,
          )
  
      # Get/transate filters
      ds_filters = None
      if filters is not None:
          ds_filters = _filters_to_expression(filters)
  
      # Define subset of `dataset_info` required by _collect_file_parts
      dataset_info_kwargs = {
          "fs": fs,
          "split_row_groups": split_row_groups,
          "gather_statistics": gather_statistics,
          "filters": filters,
          "ds_filters": ds_filters,
          "schema": schema,
          "stat_col_indices": stat_col_indices,
          "aggregation_depth": aggregation_depth,
          "blocksize": blocksize,
          "partitions": partitions,
          "dataset_options": kwargs["dataset"],
      }
  
      # Main parts/stats-construction
      if (
          has_metadata_file
          or metadata_task_size == 0
          or metadata_task_size > len(ds.files)
      ):
          # We have a global _metadata file to work with.
          # Therefore, we can just loop over fragments on the client.
  
          # Start with sorted (by path) list of file-based fragments
          file_frags = sorted(
              (frag for frag in ds.get_fragments(ds_filters)),
              key=lambda x: natural_sort_key(x.path),
          )
          parts, stats = cls._collect_file_parts(
              file_frags, dataset_info_kwargs
          )
      else:
          # We DON'T have a global _metadata file to work with.
          # We should loop over files in parallel
  
          if filters and partitions:
              # Start with sorted (by path) list of file-based fragments
              all_files = sorted(
                  (frag for frag in ds.get_fragments(ds_filters)),
                  key=lambda x: natural_sort_key(x.path),
              )
          else:
              # Collect list of file paths.
              # If valid_paths is not None, the user passed in a list
              # of files containing a _metadata file.  Since we used
              # the _metadata file to generate our dataset object , we need
              # to ignore any file fragments that are not in the list.
              all_files = sorted(ds.files, key=natural_sort_key)
              if valid_paths:
                  all_files = [
                      filef
                      for filef in all_files
                      if filef.split(fs.sep)[-1] in valid_paths
                  ]
  
          parts, stats = [], []
          if all_files:
              # Build and compute a task graph to construct stats/parts
              gather_parts_dsk = {}
              name = "gather-pq-parts-" + tokenize(
                  all_files, dataset_info_kwargs
              )
              finalize_list = []
              for task_i, file_i in enumerate(
                  range(0, len(all_files), metadata_task_size)
              ):
                  finalize_list.append((name, task_i))
                  gather_parts_dsk[finalize_list[-1]] = (
                      cls._collect_file_parts,
                      all_files[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 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 11 for calling the ArrowDatasetEngine 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