GH-39530: [Python] Allow specifying CacheOptions in ParquetFile and I…

python/pyarrow/_parquet.pyx

@@ -26,7 +26,7 @@ from cython.operator cimport dereference as deref
 from pyarrow.includes.common cimport *
 from pyarrow.includes.libarrow cimport *
 from pyarrow.includes.libarrow_python cimport *
-from pyarrow.lib cimport (_Weakrefable, Buffer, Schema,
+from pyarrow.lib cimport (_Weakrefable, CacheOptions, Buffer, Schema,
                           check_status,
                           MemoryPool, maybe_unbox_memory_pool,
                           Table, KeyValueMetadata,
@@ -1588,6 +1588,7 @@ cdef class ParquetReader(_Weakrefable):
              read_dictionary=None, binary_type=None, list_type=None,
              FileMetaData metadata=None,
              int buffer_size=0, bint pre_buffer=True,
+             CacheOptions cache_options=None,
              coerce_int96_timestamp_unit=None,
              FileDecryptionProperties decryption_properties=None,
              thrift_string_size_limit=None,
@@ -1608,6 +1609,11 @@ cdef class ParquetReader(_Weakrefable):
         metadata : FileMetaData, optional
         buffer_size : int, default 0
         pre_buffer : bool, default True
+        cache_options : pyarrow.CacheOptions, optional
+            Cache options used when pre_buffer is enabled. The default values
+            should be good for most use cases. You may want to adjust these
+            for example if you have exceptionally high latency to the file
+            system.
         coerce_int96_timestamp_unit : str, optional
         decryption_properties : FileDecryptionProperties, optional
         thrift_string_size_limit : int, optional
@@ -1660,6 +1666,9 @@ cdef class ParquetReader(_Weakrefable):
 
         arrow_props.set_pre_buffer(pre_buffer)
 
+        if cache_options is not None:
+            arrow_props.set_cache_options(cache_options.unwrap())
+
         properties.set_page_checksum_verification(page_checksum_verification)
 
         if binary_type is not None:

python/pyarrow/includes/libarrow.pxd

@@ -1860,6 +1860,7 @@ cdef extern from "arrow/ipc/api.h" namespace "arrow::ipc" nogil:
         c_bool use_threads
         c_bool ensure_native_endian
         CAlignment ensure_alignment
+        CCacheOptions pre_buffer_cache_options
 
         @staticmethod
         CIpcReadOptions Defaults()

python/pyarrow/ipc.pxi

@@ -150,20 +150,25 @@ cdef class IpcReadOptions(_Weakrefable):
         If empty (the default), return all deserialized fields.
         If non-empty, the values are the indices of fields to read on
         the top-level schema
+    pre_buffer_cache_options : pyarrow.CacheOptions, optional
+        Cache options used when pre-buffering is requested.
     """
     __slots__ = ()
 
     # cdef block is in lib.pxd
 
     def __init__(self, *, bint ensure_native_endian=True,
                  Alignment ensure_alignment=Alignment.Any,
-                 bint use_threads=True, list included_fields=None):
+                 bint use_threads=True, list included_fields=None,
+                 pre_buffer_cache_options=None):
         self.c_options = CIpcReadOptions.Defaults()
         self.ensure_native_endian = ensure_native_endian
         self.ensure_alignment = ensure_alignment
         self.use_threads = use_threads
         if included_fields is not None:
             self.included_fields = included_fields
+        if pre_buffer_cache_options is not None:
+            self.pre_buffer_cache_options = pre_buffer_cache_options
 
     @property
     def ensure_native_endian(self):
@@ -197,14 +202,23 @@ cdef class IpcReadOptions(_Weakrefable):
     def included_fields(self, list value not None):
         self.c_options.included_fields = value
 
+    @property
+    def pre_buffer_cache_options(self):
+        return CacheOptions.wrap(self.c_options.pre_buffer_cache_options)
+
+    @pre_buffer_cache_options.setter
+    def pre_buffer_cache_options(self, CacheOptions value):
+        self.c_options.pre_buffer_cache_options = value.unwrap()
+
     def __repr__(self):
         alignment = Alignment(self.ensure_alignment).name
 
         return (f"<pyarrow.ipc.IpcReadOptions "
                 f"ensure_native_endian={self.ensure_native_endian} "
                 f"ensure_alignment={alignment} "
                 f"use_threads={self.use_threads} "
-                f"included_fields={self.included_fields}>")
+                f"included_fields={self.included_fields} "
+                f"pre_buffer_cache_options={self.pre_buffer_cache_options}>")
 
 
 cdef IpcReadOptions wrap_ipc_read_options(CIpcReadOptions c):
@@ -216,6 +230,8 @@ cdef IpcReadOptions wrap_ipc_read_options(CIpcReadOptions c):
         ensure_alignment=c.ensure_alignment,
         use_threads=c.use_threads,
         included_fields=c.included_fields,
+        pre_buffer_cache_options=CacheOptions.wrap(
+            c.pre_buffer_cache_options),
     )
 
 

python/pyarrow/parquet/core.py

@@ -243,6 +243,11 @@ class ParquetFile:
         performs readahead (e.g. fsspec's S3FS), disable readahead for best
         results. Set to False if you want to prioritize minimal memory usage
         over maximum speed.
+    cache_options : pyarrow.CacheOptions, default None
+        Cache options used when pre_buffer is enabled. The default values
+        should be good for most use cases. You may want to adjust these
+        for example if you have exceptionally high latency to the file
+        system.
     coerce_int96_timestamp_unit : str, default None
         Cast timestamps that are stored in INT96 format to a particular
         resolution (e.g. 'ms'). Setting to None is equivalent to 'ns'
@@ -321,6 +326,7 @@ class ParquetFile:
     def __init__(self, source, *, metadata=None, common_metadata=None,
                  read_dictionary=None, binary_type=None, list_type=None,
                  memory_map=False, buffer_size=0, pre_buffer=True,
+                 cache_options=None,
                  coerce_int96_timestamp_unit=None,
                  decryption_properties=None, thrift_string_size_limit=None,
                  thrift_container_size_limit=None, schema_depth_limit=None,
@@ -340,6 +346,7 @@ def __init__(self, source, *, metadata=None, common_metadata=None,
         self.reader.open(
             source, use_memory_map=memory_map,
             buffer_size=buffer_size, pre_buffer=pre_buffer,
+            cache_options=cache_options,
             read_dictionary=read_dictionary, metadata=metadata,
             binary_type=binary_type, list_type=list_type,
             coerce_int96_timestamp_unit=coerce_int96_timestamp_unit,