docstr: Use parquet schema and hive-partition information
  (stored in dataset_info) to construct DataFrame metadata.
docstr_tok: ["Use","parquet","schema","and","hive-partition","information","(","stored","in","dataset_info",")","to","construct","DataFrame","metadata","."]
code: def _create_dd_meta(cls, dataset_info):
      """Use parquet schema and hive-partition information
      (stored in dataset_info) to construct DataFrame metadata.
      """
  
      # Collect necessary information from dataset_info
      schema = dataset_info["schema"]
      index = dataset_info["index"]
      categories = dataset_info["categories"]
      partition_obj = dataset_info["partitions"]
      partitions = dataset_info["partition_names"]
      physical_column_names = dataset_info.get(
          "physical_schema", schema
      ).names
      columns = None
  
      # Use pandas metadata to update categories
      pandas_metadata = _get_pandas_metadata(schema) or {}
      if pandas_metadata:
          if categories is None:
              categories = []
              for col in pandas_metadata["columns"]:
                  if (col["pandas_type"] == "categorical") and (
                      col["name"] not in categories
                  ):
                      categories.append(col["name"])
  
      # Use _arrow_table_to_pandas to generate meta
      arrow_to_pandas = (
          dataset_info["kwargs"].get("arrow_to_pandas", {}).copy()
      )
      convert_string = dataset_info["kwargs"]["convert_string"]
      dtype_backend = dataset_info["kwargs"]["dtype_backend"]
      meta = cls._arrow_table_to_pandas(
          schema.empty_table(),
          categories,
          arrow_to_pandas=arrow_to_pandas,
          dtype_backend=dtype_backend,
          convert_string=convert_string,
      )
      index_names = list(meta.index.names)
      column_names = list(meta.columns)
      if index_names and index_names != [None]:
          # Reset the index if non-null index name
          meta.reset_index(inplace=True)
  
      # Use index specified in the pandas metadata if
      # the index column was not specified by the user
      if (
          index is None
          and index_names
          and (
              # Only set to `[None]` if pandas metadata includes an index
              index_names != [None]
              or pandas_metadata.get("index_columns", None)
          )
      ):
          index = index_names
  
      # Set proper index for meta
      index_cols = index or ()
      if index_cols and index_cols != [None]:
          meta.set_index(index_cols, inplace=True)
  
      # Ensure that there is no overlap between partition columns
      # and explicit column storage
      if partitions:
          _partitions = [
              p for p in partitions if p not in physical_column_names
          ]
          if not _partitions:
              partitions = []
              dataset_info["partitions"] = []
              dataset_info["partition_keys"] = {}
              dataset_info["partition_names"] = partitions
          elif len(_partitions) != len(partitions):
              raise ValueError(
                  "No partition-columns should be written in the \n"
                  "file unless they are ALL written in the file.\n"
                  "physical columns: {} | partitions: {}".format(
                      physical_column_names, partitions
                  )
              )
  
      # Get all available column names
      column_names, index_names = _normalize_index_columns(
          columns, column_names + partitions, index, index_names
      )
      all_columns = index_names + column_names
  
      if categories:
          # Check that categories are included in columns
          if not set(categories).intersection(all_columns):
              raise ValueError(
                  "categories not in available columns.\n"
                  "categories: {} | columns: {}".format(
                      categories, list(all_columns)
                  )
              )
  
          # Make sure all categories are set to "unknown".
          # Cannot include index names in the `cols` argument.
          meta = clear_known_categories(
              meta,
              cols=[c for c in categories if c not in meta.index.names],
              dtype_backend=dtype_backend,
          )
  
      if partition_obj:
          # Update meta dtypes for partitioned columns
          for partition in partition_obj:
              if not len(partition.keys):
                  continue
              if isinstance(index, list) and partition.name == index[0]:
                  # Index from directory structure
                  meta.index = pd.CategoricalIndex(
                      [], categories=partition.keys, name=index[0]
                  )
              elif partition.name == meta.index.name:
                  # Index created from a categorical column
                  meta.index = pd.CategoricalIndex(
                      [],
                      categories=partition.keys,
                      name=meta.index.name,
                  )
              elif partition.name in meta.columns:
                  meta[partition.name] = pd.Series(
                      pd.Categorical(
                          categories=partition.keys, values=[]
                      ),
                      index=meta.index,
                  )
  
      # Update `dataset_info` and return `meta`
      dataset_info["index"] = index
      dataset_info["index_cols"] = index_cols
      dataset_info["categories"] = categories
  
      return meta
  
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 10 for calling the ArrowDatasetEngine obj's underlying member method code with example usage: obj._create_dd_meta(cls, dataset_info) and returns: meta