code: def get_read_tasks(self, parallelism: int) -> List[ReadTask]:
      import numpy as np
  
      open_stream_args = self._open_stream_args
      reader_args = self._reader_args
      _block_udf = self._block_udf
  
      paths, file_sizes = self._paths, self._file_sizes
      if self._partition_filter is not None:
          raise ValueError(
              "partition_filter is not currently supported by this class"
          )
  
      read_stream = self._delegate._read_stream
      filesystem = _wrap_s3_serialization_workaround(self._filesystem)
  
      if open_stream_args is None:
          open_stream_args = {}
  
      open_input_source = self._delegate._open_input_source
  
      def read_files(
          read_paths_and_idxs: List[Tuple[str, int]],
          fs: Union["pyarrow.fs.FileSystem", _S3FileSystemWrapper],
      ) -> Iterable[Block]:
          logger.debug(f"Reading {len(read_paths)} files.")
          if isinstance(fs, _S3FileSystemWrapper):
              fs = fs.unwrap()
          ctx = DatasetContext.get_current()
          output_buffer = BlockOutputBuffer(
              block_udf=_block_udf,
              target_max_block_size=ctx.target_max_block_size,
          )
          for read_path_and_idx in read_paths_and_idxs:
              read_path, _ = read_path_and_idx
              # Get reader_args and open_stream_args only if valid path.
              if read_path is not None:
                  compression = open_stream_args.pop(
                      "compression", None
                  )
                  if compression is None:
                      import pyarrow as pa
  
                      try:
                          # If no compression manually given, try to detect
                          # compression codec from path.
                          compression = pa.Codec.detect(read_path).name
                      except (ValueError, TypeError):
                          # Arrow's compression inference on the file path
                          # doesn't work for Snappy, so we double-check ourselves.
                          import pathlib
  
                          suffix = pathlib.Path(read_path).suffix
                          if suffix and suffix[1:] == "snappy":
                              compression = "snappy"
                          else:
                              compression = None
                  if compression == "snappy":
                      # Pass Snappy compression as a reader arg, so datasource subclasses
                      # can manually handle streaming decompression in
                      # self._read_stream().
                      reader_args["compression"] = compression
                      reader_args["filesystem"] = fs
                  elif compression is not None:
                      # Non-Snappy compression, pass as open_input_stream() arg so Arrow
                      # can take care of streaming decompression for us.
                      open_stream_args["compression"] = compression
  
              with open_input_source(
                  fs, read_path, **open_stream_args
              ) as f:
                  for data in read_stream(
                      f, read_path_and_idx, **reader_args
                  ):
                      output_buffer.add_block(data)
                      if output_buffer.has_next():
                          yield output_buffer.next()
          output_buffer.finalize()
          if output_buffer.has_next():
              yield output_buffer.next()
  
      # fix https://github.com/ray-project/ray/issues/24296
      parallelism = min(parallelism, len(paths))
  
      read_tasks = []
      for read_paths_and_idxs in np.array_split(paths, parallelism):
          if len(read_paths_and_idxs) <= 0:
              continue
  
          read_paths, _ = zip(*read_paths_and_idxs)
          meta = self._meta_provider(
              read_paths,
              self._schema,
              rows_per_file=self._delegate._rows_per_file(),
              file_sizes=file_sizes,
          )
  
          read_task = ReadTask(
              lambda read_paths_and_idxs=read_paths_and_idxs: read_files(
                  read_paths_and_idxs, filesystem
              ),
              meta,
          )
          read_tasks.append(read_task)
  
      return read_tasks
  
imports: import contextlib
  import logging
  from typing import Any, Callable, Dict, Iterable, List, Optional, Tuple, TYPE_CHECKING, Union
  import ray
  import urllib3
  from packaging import version
  from ray.data.block import Block
  from ray.data.context import DatasetContext
  from ray.data.datasource.binary_datasource import BinaryDatasource
  from ray.data.datasource.datasource import Datasource, ReadTask
  from ray.data.datasource.file_based_datasource import _check_pyarrow_version, _resolve_paths_and_filesystem, _S3FileSystemWrapper, _wrap_s3_serialization_workaround, BaseFileMetadataProvider, BlockOutputBuffer, DefaultFileMetadataProvider
  from ludwig.utils.fs_utils import get_bytes_obj_from_http_path, is_http
label_desc: Use image node_id 3 for calling the _BinaryIgnoreNoneTypeDatasourceReader obj's underlying member method code with example usage: obj.get_read_tasks(parallelism) and returns: read_tasks