code: def _process_dataset(
      paths,
      fs,
      filters=None,
      row_groups=None,
      categorical_partitions=True,
      dataset_kwargs=None,
  ):
      # Returns:
      #     file_list - Expanded/filtered list of paths
      #     row_groups - Filtered list of row-group selections
      #     partition_keys - list of partition keys for each file
      #     partition_categories - Categories for each partition
  
      # The general purpose of this function is to (1) expand
      # directory input into a list of paths (using the pyarrow
      # dataset API), (2) to apply row-group filters, and (3)
      # to discover directory-partitioning information
  
      # Deal with case that the user passed in a directory name
      file_list = paths
      if len(paths) == 1 and ioutils.is_directory(paths[0]):
          paths = ioutils.stringify_pathlike(paths[0])
  
      # Convert filters to ds.Expression
      if filters is not None:
          from pyarrow.parquet import filters_to_expression
  
          filters = filters_to_expression(filters)
  
      # Initialize ds.FilesystemDataset
      # TODO: Remove the if len(paths) workaround after following bug is fixed:
      # https://issues.apache.org/jira/browse/ARROW-16438
      dataset = ds.dataset(
          source=paths[0] if len(paths) == 1 else paths,
          filesystem=fs,
          **(
              dataset_kwargs
              or {
                  "format": "parquet",
                  "partitioning": "hive",
              }
          ),
      )
  
      file_list = dataset.files
      if len(file_list) == 0:
          raise FileNotFoundError(
              f"{paths} could not be resolved to any files"
          )
  
      # Deal with directory partitioning
      # Get all partition keys (without filters)
      partition_categories = defaultdict(list)
      file_fragment = None
      for file_fragment in dataset.get_fragments():
          keys = ds._get_partition_keys(
              file_fragment.partition_expression
          )
          if not (keys or partition_categories):
              # Bail - This is not a directory-partitioned dataset
              break
          for k, v in keys.items():
              if v not in partition_categories[k]:
                  partition_categories[k].append(v)
          if not categorical_partitions:
              # Bail - We don't need to discover all categories.
              # We only need to save the partition keys from this
              # first `file_fragment`
              break
  
      if partition_categories and file_fragment is not None:
          # Check/correct order of `categories` using last file_frag,
          # because `_get_partition_keys` does NOT preserve the
          # partition-hierarchy order of the keys.
          cat_keys = [
              part.split("=")[0]
              for part in file_fragment.path.split(fs.sep)
              if "=" in part
          ]
          if set(partition_categories) == set(cat_keys):
              partition_categories = {
                  k: partition_categories[k]
                  for k in cat_keys
                  if k in partition_categories
              }
  
      # If we do not have partitioned data and
      # are not filtering, we can return here
      if filters is None and not partition_categories:
          return file_list, row_groups, [], {}
  
      # Record initial row_groups input
      row_groups_map = {}
      if row_groups is not None:
          # Make sure paths and row_groups map 1:1
          # and save the initial mapping
          if len(paths) != len(file_list):
              raise ValueError(
                  "Cannot specify a row_group selection for a directory path."
              )
          row_groups_map = {
              path: rgs for path, rgs in zip(paths, row_groups)
          }
  
      # Apply filters and discover partition columns
      partition_keys = []
      if partition_categories or filters is not None:
          file_list = []
          if filters is not None:
              row_groups = []
          for file_fragment in dataset.get_fragments(filter=filters):
              path = file_fragment.path
  
              # Extract hive-partition keys, and make sure they
              # are ordered the same as they are in `partition_categories`
              if partition_categories:
                  raw_keys = ds._get_partition_keys(
                      file_fragment.partition_expression
                  )
                  partition_keys.append(
                      [
                          (name, raw_keys[name])
                          for name in partition_categories.keys()
                      ]
                  )
  
              # Apply row-group filtering
              selection = row_groups_map.get(path, None)
              if selection is not None or filters is not None:
                  filtered_row_groups = [
                      rg_info.id
                      for rg_fragment in file_fragment.split_by_row_group(
                          filters,
                          schema=dataset.schema,
                      )
                      for rg_info in rg_fragment.row_groups
                  ]
              file_list.append(path)
              if filters is not None:
                  if selection is None:
                      row_groups.append(filtered_row_groups)
                  else:
                      row_groups.append(
                          [
                              rg_id
                              for rg_id in filtered_row_groups
                              if rg_id in selection
                          ]
                      )
  
      return (
          file_list,
          row_groups,
          partition_keys,
          partition_categories if categorical_partitions else {},
      )
  
imports: from __future__ import annotations
  import itertools
  import math
  import operator
  import shutil
  import tempfile
  import warnings
  from collections import defaultdict
  from contextlib import ExitStack
  from functools import partial, reduce
  from typing import Callable, Dict, List, Optional, Tuple
  from uuid import uuid4
  import numpy
  import pandas
  from pyarrow import dataset
  import cudf
  from cudf._lib import parquet
  from cudf.api.types import is_list_like
  from cudf.core.column import build_categorical_column, column_empty, full
  from cudf.utils import ioutils
  from cudf.utils.nvtx_annotation import _cudf_nvtx_annotate
label_desc: Use image node_id 4 for calling a global function with example usage: _process_dataset(paths, fs, filters, row_groups, categorical_partitions, dataset_kwargs) and returns: file_list,  row_groups,  partition_keys,  unknown, file_list,  row_groups,  list,  dict