class_docstr: File-based datasource for reading files.
  
  Don't use this class directly. Instead, subclass it and implement `_read_stream()`.
class_docstr_tok: ["File-based","datasource","for","reading","files",".","Do","n't","use","this","class","directly",".","Instead",",","subclass","it","and","implement","`","_read_stream","(",")","`","."]
docstr: Opens a source path for reading and returns the associated Arrow NativeFile.
  
  The default implementation opens the source path as a sequential input stream,
  using ctx.streaming_read_buffer_size as the buffer size if none is given by the
  caller.
  
  Implementations that do not support streaming reads (e.g. that require random
  access) should override this method.
docstr_tok: ["Opens","a","source","path","for","reading","and","returns","the","associated","Arrow","NativeFile",".","The","default","implementation","opens","the","source","path","as","a","sequential","input","stream",",","using","ctx.streaming_read_buffer_size","as","the","buffer","size","if","none","is","given","by","the","caller",".","Implementations","that","do","not","support","streaming","reads","(","e.g",".","that","require","random","access",")","should","override","this","method","."]
code: def _open_input_source(
      self,
      filesystem: "pyarrow.fs.FileSystem",
      path: str,
      **open_args,
  ) -> "pyarrow.NativeFile":
      """Opens a source path for reading and returns the associated Arrow NativeFile.
  
      The default implementation opens the source path as a sequential input stream,
      using ctx.streaming_read_buffer_size as the buffer size if none is given by the
      caller.
  
      Implementations that do not support streaming reads (e.g. that require random
      access) should override this method.
      """
      import pyarrow as pa
      from pyarrow.fs import HadoopFileSystem
  
      compression = open_args.get("compression", None)
      if compression is None:
          try:
              # If no compression manually given, try to detect
              # compression codec from path.
              compression = pa.Codec.detect(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(path).suffix
              if suffix and suffix[1:] == "snappy":
                  compression = "snappy"
              else:
                  compression = None
  
      buffer_size = open_args.pop("buffer_size", None)
      if buffer_size is None:
          ctx = DataContext.get_current()
          buffer_size = ctx.streaming_read_buffer_size
  
      if compression == "snappy":
          # Arrow doesn't support streaming Snappy decompression since the canonical
          # C++ Snappy library doesn't natively support streaming decompression. We
          # works around this by manually decompressing the file with python-snappy.
          open_args["compression"] = None
      else:
          open_args["compression"] = compression
  
      file = filesystem.open_input_stream(
          path, buffer_size=buffer_size, **open_args
      )
  
      if compression == "snappy":
          import snappy
  
          stream = io.BytesIO()
          if isinstance(filesystem, HadoopFileSystem):
              snappy.hadoop_snappy.stream_decompress(
                  src=file, dst=stream
              )
          else:
              snappy.stream_decompress(src=file, dst=stream)
          stream.seek(0)
  
          file = pa.PythonFile(stream, mode="r")
  
      return file
  
imports: import io
  import pathlib
  import posixpath
  import warnings
  from typing import TYPE_CHECKING, Any, Callable, Dict, Iterable, Iterator, List, Literal, Optional, Union
  import numpy
  import ray
  from ray._private.utils import _add_creatable_buckets_param_if_s3_uri
  from ray.data._internal.dataset_logger import DatasetLogger
  from ray.data._internal.execution.interfaces import TaskContext
  from ray.data._internal.util import _check_pyarrow_version, _is_local_scheme, call_with_retry, make_async_gen
  from ray.data.block import Block, BlockAccessor
  from ray.data.context import DataContext
  from ray.data.datasource.block_path_provider import BlockWritePathProvider
  from ray.data.datasource.datasource import Datasource, ReadTask, WriteResult
  from ray.data.datasource.file_meta_provider import BaseFileMetadataProvider, DefaultFileMetadataProvider
  from ray.data.datasource.filename_provider import FilenameProvider, _DefaultFilenameProvider
  from ray.data.datasource.partitioning import Partitioning, PathPartitionFilter, PathPartitionParser
  from ray.data.datasource.path_util import _has_file_extension, _resolve_paths_and_filesystem
  from ray.util.annotations import Deprecated, DeveloperAPI, PublicAPI
label_desc: Use image node_id 6 for calling the FileBasedDatasource obj's underlying member method code with example usage: obj._open_input_source(filesystem, path) and returns: file