diff --git a/python/pyarrow/_parquet.pyx b/python/pyarrow/_parquet.pyx index 2621c3bd6d61..ff101fbd86d7 100644 --- a/python/pyarrow/_parquet.pyx +++ b/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: diff --git a/python/pyarrow/includes/libarrow.pxd b/python/pyarrow/includes/libarrow.pxd index 296148f52816..85c26e3e28cb 100644 --- a/python/pyarrow/includes/libarrow.pxd +++ b/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() diff --git a/python/pyarrow/ipc.pxi b/python/pyarrow/ipc.pxi index 2c0d9591c12f..3231d2498837 100644 --- a/python/pyarrow/ipc.pxi +++ b/python/pyarrow/ipc.pxi @@ -150,6 +150,8 @@ 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__ = () @@ -157,13 +159,16 @@ cdef class IpcReadOptions(_Weakrefable): 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,6 +202,14 @@ 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 @@ -204,7 +217,8 @@ cdef class IpcReadOptions(_Weakrefable): 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), ) diff --git a/python/pyarrow/parquet/core.py b/python/pyarrow/parquet/core.py index fa3b897ad23a..784297318f37 100644 --- a/python/pyarrow/parquet/core.py +++ b/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,