code: def _read_paths(
      cls,
      paths,
      fs,
      columns=None,
      row_groups=None,
      filters=None,
      partitions=None,
      partitioning=None,
      partition_keys=None,
      open_file_options=None,
      dataset_kwargs=None,
      **kwargs,
  ):
      # Simplify row_groups if all None
      if row_groups == [None for path in paths]:
          row_groups = None
  
      # Make sure we read in the columns needed for row-wise
      # filtering after IO. This means that one or more columns
      # will be dropped almost immediately after IO. However,
      # we do NEED these columns for accurate filtering.
      filters = _normalize_filters(filters)
      projected_columns = None
      if columns and filters:
          projected_columns = [c for c in columns if c is not None]
          columns = sorted(
              set(v[0] for v in itertools.chain.from_iterable(filters))
              | set(projected_columns)
          )
  
      dataset_kwargs = dataset_kwargs or {}
      dataset_kwargs["partitioning"] = partitioning or "hive"
      with ExitStack() as stack:
          # Non-local filesystem handling
          paths_or_fobs = paths
          if not _is_local_filesystem(fs):
              paths_or_fobs = _open_remote_files(
                  paths_or_fobs,
                  fs,
                  context_stack=stack,
                  **_default_open_file_options(
                      open_file_options, columns, row_groups
                  ),
              )
  
          # Use cudf to read in data
          try:
              df = cudf.read_parquet(
                  paths_or_fobs,
                  engine="cudf",
                  columns=columns,
                  row_groups=row_groups if row_groups else None,
                  dataset_kwargs=dataset_kwargs,
                  categorical_partitions=False,
                  **kwargs,
              )
          except RuntimeError as err:
              # TODO: Remove try/except after null-schema issue is resolved
              # (See: https://github.com/rapidsai/cudf/issues/12702)
              if len(paths_or_fobs) > 1:
                  df = cudf.concat(
                      [
                          cudf.read_parquet(
                              pof,
                              engine="cudf",
                              columns=columns,
                              row_groups=row_groups[i]
                              if row_groups
                              else None,
                              dataset_kwargs=dataset_kwargs,
                              categorical_partitions=False,
                              **kwargs,
                          )
                          for i, pof in enumerate(paths_or_fobs)
                      ]
                  )
              else:
                  raise err
  
      # Apply filters (if any are defined)
      df = _apply_post_filters(df, filters)
  
      if projected_columns:
          # Elements of `projected_columns` may now be in the index.
          # We must filter these names from our projection
          projected_columns = [
              col
              for col in projected_columns
              if col in df._column_names
          ]
          df = df[projected_columns]
  
      if partitions and partition_keys is None:
          # Use `HivePartitioning` by default
          ds = pa_ds.dataset(
              paths,
              filesystem=fs,
              **dataset_kwargs,
          )
          frag = next(ds.get_fragments())
          if frag:
              # Extract hive-partition keys, and make sure they
              # are ordered the same as they are in `partitions`
              raw_keys = pa_ds._get_partition_keys(
                  frag.partition_expression
              )
              partition_keys = [
                  (hive_part.name, raw_keys[hive_part.name])
                  for hive_part in partitions
              ]
  
      if partition_keys:
          if partitions is None:
              raise ValueError("Must pass partition sets")
  
          for i, (name, index2) in enumerate(partition_keys):
              if len(partitions[i].keys):
                  # Build a categorical column from `codes` directly
                  # (since the category is often a larger dtype)
                  codes = as_column(
                      partitions[i].keys.get_loc(index2),
                      length=len(df),
                  )
                  df[name] = build_categorical_column(
                      categories=partitions[i].keys,
                      codes=codes,
                      size=codes.size,
                      offset=codes.offset,
                      ordered=False,
                  )
              elif name not in df.columns:
                  # Add non-categorical partition column
                  df[name] = as_column(index2, length=len(df))
  
      return df
  
imports: import itertools
  import warnings
  from contextlib import ExitStack
  from functools import partial
  from io import BufferedWriter, BytesIO, IOBase
  import numpy
  from pyarrow import dataset, parquet
  from dask import dataframe
  from dask.dataframe.io.parquet.arrow import ArrowDatasetEngine
  import cudf
  from cudf.core.column import as_column, build_categorical_column
  from cudf.io import write_to_dataset
  from cudf.io.parquet import _apply_post_filters, _default_open_file_options, _normalize_filters
  from cudf.utils.dtypes import cudf_dtype_from_pa_type
  from cudf.utils.ioutils import _ROW_GROUP_SIZE_BYTES_DEFAULT, _is_local_filesystem, _open_remote_files
label_desc: Use image node_id 3 for calling the CudfEngine obj's underlying member method code with example usage: obj._read_paths(cls, paths, fs, columns, row_groups, filters, partitions, partitioning, partition_keys, open_file_options, dataset_kwargs) and returns: df