docstr: Main worker function. Loads and returns all rows matching the predicate from a rowgroup
  
  Looks up the requested piece (a single row-group in a parquet file). If a predicate is specified,
  columns needed by the predicate are loaded first. If no rows in the rowgroup matches the predicate criteria
  the rest of the columns are not loaded.
  
  :param piece_index:
  :param shuffle_row_drop_partition: A tuple 2 of the current row drop partition and the total number
      of partitions.
  :return:
docstr_tok: ["Main","worker","function",".","Loads","and","returns","all","rows","matching","the","predicate","from","a","rowgroup","Looks","up","the","requested","piece","(","a","single","row-group","in","a","parquet","file",")",".","If","a","predicate","is","specified",",","columns","needed","by","the","predicate","are","loaded","first",".","If","no","rows","in","the","rowgroup","matches","the","predicate","criteria","the","rest","of","the","columns","are","not","loaded",".",":","param","piece_index",":",":","param","shuffle_row_drop_partition",":","A","tuple","2","of","the","current","row","drop","partition","and","the","total","number","of","partitions",".",":","return",":"]
code: def process(
      self, piece_index, worker_predicate, shuffle_row_drop_partition
  ):
      """Main worker function. Loads and returns all rows matching the predicate from a rowgroup
  
      Looks up the requested piece (a single row-group in a parquet file). If a predicate is specified,
      columns needed by the predicate are loaded first. If no rows in the rowgroup matches the predicate criteria
      the rest of the columns are not loaded.
  
      :param piece_index:
      :param shuffle_row_drop_partition: A tuple 2 of the current row drop partition and the total number
          of partitions.
      :return:
      """
  
      if not self._dataset:
          self._dataset = pq.ParquetDataset(
              self._dataset_path,
              filesystem=self._filesystem,
              validate_schema=False,
              filters=self._arrow_filters,
          )
  
      piece = self._split_pieces[piece_index]
  
      # Create pyarrow file system
      parquet_file = ParquetFile(self._dataset.fs.open(piece.path))
  
      if not isinstance(self._local_cache, NullCache):
          if worker_predicate:
              raise RuntimeError(
                  "Local cache is not supported together with predicates, "
                  "unless the dataset is partitioned by the column the predicate operates on."
              )
          if shuffle_row_drop_partition[1] != 1:
              raise RuntimeError(
                  "Local cache is not supported together with shuffle_row_drop_partitions > 1"
              )
  
      if worker_predicate:
          all_cols = self._load_rows_with_predicate(
              parquet_file,
              piece,
              worker_predicate,
              shuffle_row_drop_partition,
          )
      else:
          # Using hash of the dataset path with the relative path in order to:
          #  1. Make sure if a common cache serves multiple processes (e.g. redis), we don't have conflicts
          #  2. Dataset path is hashed, to make sure we don't create too long keys, which maybe incompatible with
          #     some cache implementations
          #  3. Still leave relative path and the piece_index in plain text to make it easier to debug
          # self._dataset_path could be a list of urls or a string.
          _dataset_path_for_hash = (
              "_".join(self._dataset_path)
              if isinstance(self._dataset_path, Iterable)
              else self._dataset_path
          )
          cache_key = "{}:{}:{}".format(
              hashlib.md5(
                  _dataset_path_for_hash.encode("utf-8")
              ).hexdigest(),
              piece.path,
              piece_index,
          )
          all_cols = self._local_cache.get(
              cache_key,
              lambda: self._load_rows(
                  parquet_file, piece, shuffle_row_drop_partition
              ),
          )
  
      if self._ngram:
          all_cols = self._ngram.form_ngram(
              data=all_cols, schema=self._schema
          )
  
      if all_cols:
          self.publish_func(all_cols)
  
imports: from __future__ import division
  import hashlib
  import threading
  from collections.abc import Iterable
  import numpy
  from pyarrow import parquet
  from pyarrow.parquet import ParquetFile
  from petastorm import utils
  from petastorm.cache import NullCache
  from petastorm.workers_pool import EmptyResultError
  from petastorm.workers_pool.worker_base import WorkerBase
label_desc: Use image node_id 3 for calling the PyDictReaderWorker obj's underlying member method code with example usage: obj.process(piece_index, worker_predicate, shuffle_row_drop_partition) without return types