code: def read_next(self, workers_pool, schema, ngram):
      try:
          assert (
              not ngram
          ), "ArrowReader does not support ngrams for now"
  
          result_table = workers_pool.get_results()
  
          # Convert arrow table columns into numpy. Strings are handled differently since to_pandas() returns
          # numpy array of dtype=object.
          result_dict = dict()
          for column_name in result_table.column_names:
              column = result_table.column(column_name)
              # Assume we get only one chunk since reader worker reads one rowgroup at a time
  
              # `to_pandas` works slower when called on the entire `data` rather directly on a chunk.
              if result_table.column(0).num_chunks == 1:
                  column_as_pandas = column.chunks[0].to_pandas()
              else:
                  column_as_pandas = column.to_pandas()
  
              # pyarrow < 0.15.0 would always return a numpy array. Starting 0.15 we get pandas series, hence we
              # convert it into numpy array
              if isinstance(column_as_pandas, pd.Series):
                  column_as_numpy = column_as_pandas.values
              else:
                  column_as_numpy = column_as_pandas
  
              if pa.types.is_string(column.type):
                  result_dict[column_name] = column_as_numpy.astype(
                      np.unicode_
                  )
              elif pa.types.is_list(column.type):
                  # Assuming all lists are of the same length, hence we can collate them into a matrix
                  list_of_lists = column_as_numpy
                  try:
                      col_data = np.vstack(list_of_lists.tolist())
                      shape = schema.fields[column_name].shape
                      if len(shape) > 1:
                          col_data = col_data.reshape(
                              (len(list_of_lists),) + shape
                          )
                      result_dict[column_name] = col_data
  
                  except ValueError:
                      raise RuntimeError(
                          "Length of all values in column '{}' are expected to be the same length. "
                          "Got the following set of lengths: '{}'".format(
                              column_name,
                              ", ".join(
                                  str(value.shape[0])
                                  for value in list_of_lists
                              ),
                          )
                      )
              else:
                  result_dict[column_name] = column_as_numpy
  
          return schema.make_namedtuple(**result_dict)
  
      except EmptyResultError:
          raise StopIteration
  
imports: from __future__ import division
  import hashlib
  import operator
  import numpy
  import pandas
  import pyarrow
  from pyarrow import parquet
  from pyarrow.parquet import ParquetFile
  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 ArrowReaderWorkerResultsQueueReader obj's underlying member method code with example usage: obj.read_next(workers_pool, schema, ngram) and returns: schema