class_docstr: Parquet datasource, for reading and writing Parquet files.
  
  The primary difference from ParquetBaseDatasource is that this uses
  PyArrow's `ParquetDataset` abstraction for dataset reads, and thus offers
  automatic Arrow dataset schema inference and row count collection at the
  cost of some potential performance and/or compatibility penalties.
class_docstr_tok: ["Parquet","datasource",",","for","reading","and","writing","Parquet","files",".","The","primary","difference","from","ParquetBaseDatasource","is","that","this","uses","PyArrow","'s","`","ParquetDataset","`","abstraction","for","dataset","reads",",","and","thus","offers","automatic","Arrow","dataset","schema","inference","and","row","count","collection","at","the","cost","of","some","potential","performance","and\/or","compatibility","penalties","."]
code: def get_read_tasks(self, parallelism: int) -> List[ReadTask]:
      # NOTE: We override the base class FileBasedDatasource.get_read_tasks()
      # method in order to leverage pyarrow's ParquetDataset abstraction,
      # which simplifies partitioning logic. We still use
      # FileBasedDatasource's write side, however.
      pq_metadata = self._metadata
      if len(pq_metadata) < len(self._pq_fragments):
          # Pad `pq_metadata` to be same length of `self._pq_fragments`.
          # This can happen when no file metadata being prefetched.
          pq_metadata += [None] * (
              len(self._pq_fragments) - len(pq_metadata)
          )
  
      if self._file_metadata_shuffler is not None:
          files_metadata = list(
              zip(self._pq_fragments, self._pq_paths, pq_metadata)
          )
          shuffled_files_metadata = [
              files_metadata[i]
              for i in self._file_metadata_shuffler.permutation(
                  len(files_metadata)
              )
          ]
          pq_fragments, pq_paths, pq_metadata = list(
              map(list, zip(*shuffled_files_metadata))
          )
      else:
          pq_fragments, pq_paths, pq_metadata = (
              self._pq_fragments,
              self._pq_paths,
              pq_metadata,
          )
  
      read_tasks = []
      for fragments, paths, metadata in zip(
          np.array_split(pq_fragments, parallelism),
          np.array_split(pq_paths, parallelism),
          np.array_split(pq_metadata, parallelism),
      ):
          if len(fragments) <= 0:
              continue
  
          meta = self._meta_provider(
              paths,
              self._inferred_schema,
              num_fragments=len(fragments),
              prefetched_metadata=metadata,
          )
          # If there is a filter operation, reset the calculated row count,
          # since the resulting row count is unknown.
          if self._to_batches_kwargs.get("filter") is not None:
              meta.num_rows = None
  
          if meta.size_bytes is not None:
              meta.size_bytes = int(
                  meta.size_bytes * self._encoding_ratio
              )
  
          if meta.num_rows and meta.size_bytes:
              # Make sure the batches read are small enough to enable yielding of
              # output blocks incrementally during the read.
              row_size = meta.size_bytes / meta.num_rows
              # Make sure the row batch size is small enough that block splitting
              # is still effective.
              max_parquet_reader_row_batch_size_bytes = (
                  DataContext.get_current().target_max_block_size // 10
              )
              default_read_batch_size_rows = max(
                  1,
                  min(
                      PARQUET_READER_ROW_BATCH_SIZE,
                      max_parquet_reader_row_batch_size_bytes
                      // row_size,
                  ),
              )
          else:
              default_read_batch_size_rows = (
                  PARQUET_READER_ROW_BATCH_SIZE
              )
          (
              block_udf,
              to_batches_kwargs,
              columns,
              schema,
              include_paths,
          ) = (
              self._block_udf,
              self._to_batches_kwargs,
              self._columns,
              self._schema,
              self._include_paths,
          )
          read_tasks.append(
              ReadTask(
                  lambda f=fragments: _read_fragments(
                      block_udf,
                      to_batches_kwargs,
                      default_read_batch_size_rows,
                      columns,
                      schema,
                      f,
                      include_paths,
                  ),
                  meta,
              )
          )
  
      return read_tasks
  
imports: import logging
  from typing import TYPE_CHECKING, Any, Callable, Dict, Iterator, List, Literal, Optional, Union
  import numpy
  import ray
  import ray.cloudpickle
  from ray.data._internal.progress_bar import ProgressBar
  from ray.data._internal.remote_fn import cached_remote_fn
  from ray.data._internal.util import _check_pyarrow_version, _is_local_scheme
  from ray.data.block import Block
  from ray.data.context import DataContext
  from ray.data.datasource import Datasource
  from ray.data.datasource._default_metadata_providers import get_generic_metadata_provider
  from ray.data.datasource.datasource import ReadTask
  from ray.data.datasource.file_meta_provider import DefaultParquetMetadataProvider, ParquetMetadataProvider, _handle_read_os_error
  from ray.data.datasource.partitioning import PathPartitionFilter
  from ray.data.datasource.path_util import _has_file_extension, _resolve_paths_and_filesystem
  from ray.util.annotations import PublicAPI
label_desc: Use image node_id 3 for calling the ParquetDatasource obj's underlying member method code with example usage: obj.get_read_tasks(parallelism) and returns: read_tasks