code: def read_parquet(
      path,
      columns=None,
      filters=None,
      categories=None,
      index=None,
      storage_options=None,
      engine="auto",
      use_nullable_dtypes: bool | None = None,
      dtype_backend=None,
      calculate_divisions=None,
      ignore_metadata_file=False,
      metadata_task_size=None,
      split_row_groups="infer",
      blocksize="default",
      aggregate_files=None,
      parquet_file_extension=(".parq", ".parquet", ".pq"),
      filesystem=None,
      **kwargs,
  ):
      """
      Read a Parquet file into a Dask DataFrame
  
      This reads a directory of Parquet data into a Dask.dataframe, one file per
      partition.  It selects the index among the sorted columns if any exist.
  
      Parameters
      ----------
      path : str or list
          Source directory for data, or path(s) to individual parquet files.
          Prefix with a protocol like ``s3://`` to read from alternative
          filesystems. To read from multiple files you can pass a globstring or a
          list of paths, with the caveat that they must all have the same
          protocol.
      columns : str or list, default None
          Field name(s) to read in as columns in the output. By default all
          non-index fields will be read (as determined by the pandas parquet
          metadata, if present). Provide a single field name instead of a list to
          read in the data as a Series.
      filters : Union[List[Tuple[str, str, Any]], List[List[Tuple[str, str, Any]]]], default None
          List of filters to apply, like ``[[('col1', '==', 0), ...], ...]``.
          Using this argument will result in row-wise filtering of the final partitions.
  
          Predicates can be expressed in disjunctive normal form (DNF). This means that
          the inner-most tuple describes a single column predicate. These inner predicates
          are combined with an AND conjunction into a larger predicate. The outer-most
          list then combines all of the combined filters with an OR disjunction.
  
          Predicates can also be expressed as a ``List[Tuple]``. These are evaluated
          as an AND conjunction. To express OR in predicates, one must use the
          (preferred for "pyarrow") ``List[List[Tuple]]`` notation.
      index : str, list or False, default None
          Field name(s) to use as the output frame index. By default will be
          inferred from the pandas parquet file metadata, if present. Use ``False``
          to read all fields as columns.
      categories : list or dict, default None
          For any fields listed here, if the parquet encoding is Dictionary,
          the column will be created with dtype category. Use only if it is
          guaranteed that the column is encoded as dictionary in all row-groups.
          If a list, assumes up to 2**16-1 labels; if a dict, specify the number
          of labels expected; if None, will load categories automatically for
          data written by dask, not otherwise.
      storage_options : dict, default None
          Key/value pairs to be passed on to the file-system backend, if any.
          Note that the default file-system backend can be configured with the
          ``filesystem`` argument, described below.
      open_file_options : dict, default None
          Key/value arguments to be passed along to ``AbstractFileSystem.open``
          when each parquet data file is open for reading. Experimental
          (optimized) "precaching" for remote file systems (e.g. S3, GCS) can
          be enabled by adding ``{"method": "parquet"}`` under the
          ``"precache_options"`` key. Also, a custom file-open function can be
          used (instead of ``AbstractFileSystem.open``), by specifying the
          desired function under the ``"open_file_func"`` key.
      engine : {'auto', 'pyarrow'}
          Parquet library to use. Defaults to 'auto', which uses ``pyarrow`` if
          it is installed, and falls back to the deprecated ``fastparquet`` otherwise.
          Note that ``fastparquet`` does not support all functionality offered by
          ``pyarrow``.
          This is also used by third-party packages (e.g. CuDF) to inject bespoke engines.
      use_nullable_dtypes : {False, True}
          Whether to use extension dtypes for the resulting ``DataFrame``.
  
          .. note::
  
              This option is deprecated. Use "dtype_backend" instead.
  
      dtype_backend : {'numpy_nullable', 'pyarrow'}, defaults to NumPy backed DataFrames
          Which dtype_backend to use, e.g. whether a DataFrame should have NumPy arrays,
          nullable dtypes are used for all dtypes that have a nullable implementation
          when 'numpy_nullable' is set, pyarrow is used for all dtypes if 'pyarrow'
          is set.
          ``dtype_backend="pyarrow"`` requires ``pandas`` 1.5+.
      calculate_divisions : bool, default False
          Whether to use min/max statistics from the footer metadata (or global
          ``_metadata`` file) to calculate divisions for the output DataFrame
          collection. Divisions will not be calculated if statistics are missing.
          This option will be ignored if ``index`` is not specified and there is
          no physical index column specified in the custom "pandas" Parquet
          metadata. Note that ``calculate_divisions=True`` may be extremely slow
          when no global ``_metadata`` file is present, especially when reading
          from remote storage. Set this to ``True`` only when known divisions
          are needed for your workload (see :ref:`dataframe-design-partitions`).
      ignore_metadata_file : bool, default False
          Whether to ignore the global ``_metadata`` file (when one is present).
          If ``True``, or if the global ``_metadata`` file is missing, the parquet
          metadata may be gathered and processed in parallel. Parallel metadata
          processing is currently supported for ``ArrowDatasetEngine`` only.
      metadata_task_size : int, default configurable
          If parquet metadata is processed in parallel (see ``ignore_metadata_file``
          description above), this argument can be used to specify the number of
          dataset files to be processed by each task in the Dask graph.  If this
          argument is set to ``0``, parallel metadata processing will be disabled.
          The default values for local and remote filesystems can be specified
          with the "metadata-task-size-local" and "metadata-task-size-remote"
          config fields, respectively (see "dataframe.parquet").
      split_row_groups : 'infer', 'adaptive', bool, or int, default 'infer'
          If True, then each output dataframe partition will correspond to a single
          parquet-file row-group. If False, each partition will correspond to a
          complete file.  If a positive integer value is given, each dataframe
          partition will correspond to that number of parquet row-groups (or fewer).
          If 'adaptive', the metadata of each file will be used to ensure that every
          partition satisfies ``blocksize``. If 'infer' (the default), the
          uncompressed storage-size metadata in the first file will be used to
          automatically set ``split_row_groups`` to either 'adaptive' or ``False``.
      blocksize : int or str, default 'default'
          The desired size of each output ``DataFrame`` partition in terms of total
          (uncompressed) parquet storage space. This argument is currenlty used to
          set the default value of ``split_row_groups`` (using row-group metadata
          from a single file), and will be ignored if ``split_row_groups`` is not
          set to 'infer' or 'adaptive'. Default is 256 MiB.
      aggregate_files : bool or str, default None
          WARNING: Passing a string argument to ``aggregate_files`` will result
          in experimental behavior. This behavior may change in the future.
  
          Whether distinct file paths may be aggregated into the same output
          partition. This parameter is only used when `split_row_groups` is set to
          'infer', 'adaptive' or to an integer >1. A setting of True means that any
          two file paths may be aggregated into the same output partition, while
          False means that inter-file aggregation is prohibited.
  
          For "hive-partitioned" datasets, a "partition"-column name can also be
          specified. In this case, we allow the aggregation of any two files
          sharing a file path up to, and including, the corresponding directory name.
          For example, if ``aggregate_files`` is set to ``"section"`` for the
          directory structure below, ``03.parquet`` and ``04.parquet`` may be
          aggregated together, but ``01.parquet`` and ``02.parquet`` cannot be.
          If, however, ``aggregate_files`` is set to ``"region"``, ``01.parquet``
          may be aggregated with ``02.parquet``, and ``03.parquet`` may be aggregated
          with ``04.parquet``::
  
              dataset-path/
              ├── region=1/
              │   ├── section=a/
              │   │   └── 01.parquet
              │   ├── section=b/
              │   └── └── 02.parquet
              └── region=2/
                  ├── section=a/
                  │   ├── 03.parquet
                  └── └── 04.parquet
  
          Note that the default behavior of ``aggregate_files`` is ``False``.
      parquet_file_extension: str, tuple[str], or None, default (".parq", ".parquet", ".pq")
          A file extension or an iterable of extensions to use when discovering
          parquet files in a directory. Files that don't match these extensions
          will be ignored. This argument only applies when ``paths`` corresponds
          to a directory and no ``_metadata`` file is present (or
          ``ignore_metadata_file=True``). Passing in ``parquet_file_extension=None``
          will treat all files in the directory as parquet files.
  
          The purpose of this argument is to ensure that the engine will ignore
          unsupported metadata files (like Spark's '_SUCCESS' and 'crc' files).
          It may be necessary to change this argument if the data files in your
          parquet dataset do not end in ".parq", ".parquet", or ".pq".
      filesystem: "fsspec", "arrow", or fsspec.AbstractFileSystem backend to use.
      dataset: dict, default None
          Dictionary of options to use when creating a ``pyarrow.dataset.Dataset`` object.
          These options may include a "filesystem" key to configure the desired
          file-system backend. However, the top-level ``filesystem`` argument will always
          take precedence.
  
          **Note**: The ``dataset`` options may include a "partitioning" key.
          However, since ``pyarrow.dataset.Partitioning``
          objects cannot be serialized, the value can be a dict of key-word
          arguments for the ``pyarrow.dataset.partitioning`` API
          (e.g. ``dataset={"partitioning": {"flavor": "hive", "schema": ...}}``).
          Note that partitioned columns will not be converted to categorical
          dtypes when a custom partitioning schema is specified in this way.
      read: dict, default None
          Dictionary of options to pass through to ``engine.read_partitions``
          using the ``read`` key-word argument.
      arrow_to_pandas: dict, default None
          Dictionary of options to use when converting from ``pyarrow.Table`` to
          a pandas ``DataFrame`` object. Only used by the "arrow" engine.
      **kwargs: dict (of dicts)
          Options to pass through to ``engine.read_partitions`` as stand-alone
          key-word arguments. Note that these options will be ignored by the
          engines defined in ``dask.dataframe``, but may be used by other custom
          implementations.
  
      Examples
      --------
      >>> df = dd.read_parquet('s3://bucket/my-parquet-data')  # doctest: +SKIP
  
      See Also
      --------
      to_parquet
      pyarrow.parquet.ParquetDataset
      """
      # Handle `chunksize` deprecation
      if "chunksize" in kwargs:
          if blocksize != "default":
              raise ValueError(
                  "The `chunksize` argument is now deprecated. "
                  "Please use the `blocksize` argument instead."
              )
          blocksize = kwargs.pop("chunksize")
          warnings.warn(
              "The `chunksize` argument is deprecated, and will be "
              "removed in a future release. Setting the `blocksize` "
              "argument instead. Please see documentation on the "
              "`blocksize` argument for more information.",
              FutureWarning,
          )
  
      # FutureWarning for `aggregate_files`
      if aggregate_files and isinstance(aggregate_files, str):
          warnings.warn(
              "String support for `aggregate_files` is experimental. "
              "Behavior may change in the future. ",
              FutureWarning,
          )
  
      if "read_from_paths" in kwargs:
          kwargs.pop("read_from_paths")
          warnings.warn(
              "`read_from_paths` is no longer supported and will be ignored.",
              FutureWarning,
          )
  
      # Handle gather_statistics deprecation
      if "gather_statistics" in kwargs:
          if calculate_divisions is None:
              calculate_divisions = kwargs.pop("gather_statistics")
              warnings.warn(
                  "``gather_statistics`` is deprecated and will be removed in a "
                  "future release. Please use ``calculate_divisions`` instead.",
                  FutureWarning,
              )
          else:
              warnings.warn(
                  f"``gather_statistics`` is deprecated. Ignoring this option "
                  f"in favor of ``calculate_divisions={calculate_divisions}``",
                  FutureWarning,
              )
      calculate_divisions = bool(calculate_divisions)
  
      # We support a top-level `parquet_file_extension` kwarg, but
      # must check if the deprecated `require_extension` option is
      # being passed to the engine. If `parquet_file_extension` is
      # set to the default value, and `require_extension` was also
      # specified, we will use `require_extension` but warn the user.
      if (
          "dataset" in kwargs
          and "require_extension" in kwargs["dataset"]
          and parquet_file_extension == (".parq", ".parquet", ".pq")
      ):
          parquet_file_extension = kwargs["dataset"].pop(
              "require_extension"
          )
          warnings.warn(
              "require_extension is deprecated, and will be removed from "
              "read_parquet in a future release. Please use the top-level "
              "parquet_file_extension argument instead.",
              FutureWarning,
          )
  
      # Store initial function arguments
      input_kwargs = {
          "columns": columns,
          "filters": filters,
          "categories": categories,
          "index": index,
          "storage_options": storage_options,
          "engine": engine,
          "use_nullable_dtypes": use_nullable_dtypes,
          "dtype_backend": dtype_backend,
          "calculate_divisions": calculate_divisions,
          "ignore_metadata_file": ignore_metadata_file,
          "metadata_task_size": metadata_task_size,
          "split_row_groups": split_row_groups,
          "blocksize": blocksize,
          "aggregate_files": aggregate_files,
          "parquet_file_extension": parquet_file_extension,
          "filesystem": filesystem,
          **kwargs,
      }
  
      if isinstance(columns, str):
          input_kwargs["columns"] = [columns]
          df = read_parquet(path, **input_kwargs)
          return df[columns]
  
      if columns is not None:
          columns = list(columns)
  
      engine = get_engine(engine)
  
      if hasattr(path, "name"):
          path = stringify_path(path)
  
      # Update input_kwargs
      input_kwargs.update({"columns": columns, "engine": engine})
  
      # Process and split user options
      (
          dataset_options,
          read_options,
          open_file_options,
          other_options,
      ) = _split_user_options(**kwargs)
  
      # Extract global filesystem and paths
      (
          fs,
          paths,
          dataset_options,
          open_file_options,
      ) = engine.extract_filesystem(
          path,
          filesystem,
          dataset_options,
          open_file_options,
          storage_options,
      )
      read_options["open_file_options"] = open_file_options
      paths = sorted(
          paths, key=natural_sort_key
      )  # numeric rather than glob ordering
  
      auto_index_allowed = False
      if index is None:
          # User is allowing auto-detected index
          auto_index_allowed = True
      if index and isinstance(index, str):
          index = [index]
  
      if split_row_groups in ("infer", "adaptive"):
          # Using blocksize to plan partitioning
          blocksize = (
              engine.default_blocksize()
              if blocksize == "default"
              else blocksize
          )
      else:
          # Not using blocksize - Set to `None`
          blocksize = None
  
      read_metadata_result = engine.read_metadata(
          fs,
          paths,
          categories=categories,
          index=index,
          use_nullable_dtypes=use_nullable_dtypes,
          dtype_backend=dtype_backend,
          gather_statistics=calculate_divisions,
          filters=filters,
          split_row_groups=split_row_groups,
          blocksize=blocksize,
          aggregate_files=aggregate_files,
          ignore_metadata_file=ignore_metadata_file,
          metadata_task_size=metadata_task_size,
          parquet_file_extension=parquet_file_extension,
          dataset=dataset_options,
          read=read_options,
          **other_options,
      )
  
      # In the future, we may want to give the engine the
      # option to return a dedicated element for `common_kwargs`.
      # However, to avoid breaking the API, we just embed this
      # data in the first element of `parts` for now.
      # The logic below is inteded to handle backward and forward
      # compatibility with a user-defined engine.
      meta, statistics, parts, index = read_metadata_result[:4]
      common_kwargs = {}
      aggregation_depth = False
      if len(parts):
          # For now, `common_kwargs` and `aggregation_depth`
          # may be stored in the first element of `parts`
          common_kwargs = parts[0].pop("common_kwargs", {})
          aggregation_depth = parts[0].pop(
              "aggregation_depth", aggregation_depth
          )
          split_row_groups = parts[0].pop(
              "split_row_groups", split_row_groups
          )
  
      # Parse dataset statistics from metadata (if available)
      parts, divisions, index = process_statistics(
          parts,
          statistics,
          filters,
          index,
          (blocksize if split_row_groups is True else None),
          split_row_groups,
          fs,
          aggregation_depth,
      )
  
      # Account for index and columns arguments.
      # Modify `meta` dataframe accordingly
      meta, index, columns = set_index_columns(
          meta, index, columns, auto_index_allowed
      )
      if meta.index.name == NONE_LABEL:
          meta.index.name = None
  
      if len(divisions) < 2:
          # empty dataframe - just use meta
          divisions = (None, None)
          io_func = lambda x: x
          parts = [meta]
      else:
          # Use IO function wrapper
          io_func = ParquetFunctionWrapper(
              engine,
              fs,
              meta,
              columns,
              index,
              dtype_backend,
              {},  # All kwargs should now be in `common_kwargs`
              common_kwargs,
          )
  
      # If we are using a remote filesystem and retries is not set, bump it
      # to be more fault tolerant, as transient transport errors can occur.
      # The specific number 5 isn't hugely motivated: it's less than ten and more
      # than two.
      annotations = dask.get_annotations()
      if "retries" not in annotations and not _is_local_fs(fs):
          ctx = dask.annotate(retries=5)
      else:
          ctx = contextlib.nullcontext()  # type: ignore
  
      with ctx:
          # Construct the output collection with from_map
          return from_map(
              io_func,
              parts,
              meta=meta,
              divisions=divisions,
              label="read-parquet",
              token=tokenize(
                  path,
                  dask.config.get("dataframe.convert-string"),
                  **input_kwargs,
              ),
              enforce_metadata=False,
              creation_info={
                  "func": read_parquet,
                  "args": (path,),
                  "kwargs": input_kwargs,
              },
          )
  
imports: from __future__ import annotations
  import contextlib
  import math
  import warnings
  from typing import Literal
  import pandas
  import tlz
  from fsspec.core import get_fs_token_paths
  from fsspec.utils import stringify_path
  import dask
  from dask.base import tokenize
  from dask.blockwise import BlockIndex
  from dask.dataframe.backends import dataframe_creation_dispatch
  from dask.dataframe.core import DataFrame, Scalar
  from dask.dataframe.io.io import from_map
  from dask.dataframe.io.parquet.utils import Engine, _sort_and_analyze_paths, _split_user_options
  from dask.dataframe.io.utils import DataFrameIOFunction, _is_local_fs
  from dask.dataframe.methods import concat
  from dask.delayed import Delayed
  from dask.highlevelgraph import HighLevelGraph
  from dask.layers import DataFrameIOLayer
  from dask.utils import apply, import_required, natural_sort_key, parse_bytes
label_desc: Use image node_id 1 for calling a global function with example usage: read_parquet(path, columns, filters, categories, index, storage_options, engine, use_nullable_dtypes, dtype_backend, calculate_divisions, ignore_metadata_file, metadata_task_size, split_row_groups, blocksize, aggregate_files, parquet_file_extension, filesystem) and returns: df, from_map