mirrors/arrow
0
0 Fork 0
mirror of https://github.com/apache/arrow.git synced 2026-10-29 17:15:39 +08:00
Code Activity

s3 filesystem bindings

Browse Source
This commit is contained in:
Krisztián Szűcs committed 2019-09-24 16:49:03 +02:00
1 parent c6faaed9a7
commit 5200af16fd
24 files changed
+345 -230
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

cpp/cmake_modules/ThirdpartyToolchain.cmake
+6
View File
@@ -2535,6 +2535,12 @@ if(ARROW_S3)
include_directories(SYSTEM ${AWSSDK_INCLUDE_DIR})
message(STATUS "Found AWS SDK headers: ${AWSSDK_INCLUDE_DIR}")
message(STATUS "Found AWS SDK libraries: ${AWSSDK_LINK_LIBRARIES}")
if(APPLE)
set_target_properties(AWS::aws-c-common PROPERTIES
INTERFACE_LINK_LIBRARIES "-pthread;pthread;-framework CoreFoundation"
)
endif()
endif()
# Write out the package configurations.
python/CMakeLists.txt
+4
View File
@@ -385,6 +385,10 @@ set(CYTHON_EXTENSIONS lib _fs _csv _json)
set(LINK_LIBS arrow_shared arrow_python_shared)
if(PYARROW_BUILD_S3)
set(CYTHON_EXTENSIONS ${CYTHON_EXTENSIONS} _s3)
endif()
if(PYARROW_BUILD_CUDA)
# Arrow CUDA
find_package(ArrowCuda)
python/pyarrow/_csv.pyx
+2 -2
View File
@@ -466,7 +466,7 @@ cdef class ConvertOptions:
self.options.include_missing_columns = value
cdef _get_reader(input_file, shared_ptr[InputStream]* out):
cdef _get_reader(input_file, shared_ptr[CInputStream]* out):
use_memory_map = False
get_input_stream(input_file, use_memory_map, out)
@@ -522,7 +522,7 @@ def read_csv(input_file, read_options=None, parse_options=None,
Contents of the CSV file as a in-memory table.
"""
cdef:
shared_ptr[InputStream] stream
shared_ptr[CInputStream] stream
CCSVReadOptions c_read_options
CCSVParseOptions c_parse_options
CCSVConvertOptions c_convert_options
python/pyarrow/_cuda.pyx
+2 -2
View File
@@ -729,7 +729,7 @@ cdef class BufferReader(NativeFile):
self.buffer = obj
self.reader = new CCudaBufferReader(self.buffer.buffer)
self.set_random_access_file(
shared_ptr[RandomAccessFile](self.reader))
shared_ptr[CRandomAccessFile](self.reader))
self.is_readable = True
def read_buffer(self, nbytes=None):
@@ -776,7 +776,7 @@ cdef class BufferWriter(NativeFile):
def __cinit__(self, CudaBuffer buffer):
self.buffer = buffer
self.writer = new CCudaBufferWriter(self.buffer.cuda_buffer)
self.set_output_stream(shared_ptr[OutputStream](self.writer))
self.set_output_stream(shared_ptr[COutputStream](self.writer))
self.is_writable = True
def writeat(self, int64_t position, object data):
python/pyarrow/_fs.pyx
+2 -11
View File
@@ -22,17 +22,8 @@ import six
from pyarrow.compat import frombytes, tobytes
from pyarrow.includes.common cimport *
from pyarrow.includes.libarrow cimport PyDateTime_from_TimePoint
from pyarrow.includes.libarrow_fs cimport *
from pyarrow.util import _stringify_path
from pyarrow.lib import _detect_compression
from pyarrow.lib cimport (
check_status,
NativeFile,
BufferedOutputStream,
BufferedInputStream,
CompressedInputStream,
CompressedOutputStream
)
from pyarrow.lib cimport *
cdef inline c_string _path_as_bytes(path) except *:
@@ -523,4 +514,4 @@ cdef class SubTreeFileSystem(FileSystem):
cdef init(self, const shared_ptr[CFileSystem]& wrapped):
FileSystem.init(self, wrapped)
self.subtreefs = <CSubTreeFileSystem*> wrapped.get()
self.subtreefs = <CSubTreeFileSystem*> wrapped.get()
python/pyarrow/_json.pyx
+2 -2
View File
@@ -135,7 +135,7 @@ cdef class ParseOptions:
self.options.newlines_in_values = value
cdef _get_reader(input_file, shared_ptr[InputStream]* out):
cdef _get_reader(input_file, shared_ptr[CInputStream]* out):
use_memory_map = False
get_input_stream(input_file, use_memory_map, out)
@@ -175,7 +175,7 @@ def read_json(input_file, read_options=None, parse_options=None,
Contents of the JSON file as a in-memory table.
"""
cdef:
shared_ptr[InputStream] stream
shared_ptr[CInputStream] stream
CJSONReadOptions c_read_options
CJSONParseOptions c_parse_options
shared_ptr[CJSONReader] reader
python/pyarrow/_orc.pxd
+2 -2
View File
@@ -28,7 +28,7 @@ from pyarrow.includes.libarrow cimport (CArray, CSchema, CStatus,
CKeyValueMetadata,
CRecordBatch,
CTable,
RandomAccessFile, OutputStream,
RandomAccessFile, COutputStream,
TimeUnit)
@@ -37,7 +37,7 @@ cdef extern from "arrow/adapters/orc/adapter.h" \
cdef cppclass ORCFileReader:
@staticmethod
CStatus Open(const shared_ptr[RandomAccessFile]& file,
CStatus Open(const shared_ptr[CRandomAccessFile]& file,
CMemoryPool* pool,
unique_ptr[ORCFileReader]* reader)
python/pyarrow/_orc.pyx
+1 -1
View File
@@ -46,7 +46,7 @@ cdef class ORCReader:
def open(self, object source, c_bool use_memory_map=True):
cdef:
shared_ptr[RandomAccessFile] rd_handle
shared_ptr[CRandomAccessFile] rd_handle
self.source = source
python/pyarrow/_parquet.pxd
+5 -5
View File
@@ -24,7 +24,7 @@ from pyarrow.includes.common cimport *
from pyarrow.includes.libarrow cimport (CChunkedArray, CSchema, CStatus,
CTable, CMemoryPool, CBuffer,
CKeyValueMetadata,
RandomAccessFile, OutputStream,
CRandomAccessFile, COutputStream,
TimeUnit)
@@ -316,7 +316,7 @@ cdef extern from "parquet/api/reader.h" namespace "parquet" nogil:
unique_ptr[CRowGroupMetaData] RowGroup(int i)
const SchemaDescriptor* schema()
shared_ptr[const CKeyValueMetadata] key_value_metadata() const
void WriteTo(OutputStream* dst) const
void WriteTo(COutputStream* dst) const
cdef shared_ptr[CFileMetaData] CFileMetaData_Make \
" parquet::FileMetaData::Make"(const void* serialized_metadata,
@@ -406,7 +406,7 @@ cdef extern from "parquet/arrow/reader.h" namespace "parquet::arrow" nogil:
cdef cppclass FileReaderBuilder:
FileReaderBuilder()
CStatus Open(const shared_ptr[RandomAccessFile]& file,
CStatus Open(const shared_ptr[CRandomAccessFile]& file,
const CReaderProperties& properties,
const shared_ptr[CFileMetaData]& metadata)
@@ -435,7 +435,7 @@ cdef extern from "parquet/arrow/writer.h" namespace "parquet::arrow" nogil:
@staticmethod
CStatus Open(const CSchema& schema, CMemoryPool* pool,
const shared_ptr[OutputStream]& sink,
const shared_ptr[COutputStream]& sink,
const shared_ptr[WriterProperties]& properties,
const shared_ptr[ArrowWriterProperties]& arrow_properties,
unique_ptr[FileWriter]* writer)
@@ -448,4 +448,4 @@ cdef extern from "parquet/arrow/writer.h" namespace "parquet::arrow" nogil:
CStatus WriteMetaDataFile(
const CFileMetaData& file_metadata,
const OutputStream* sink)
const COutputStream* sink)
python/pyarrow/_parquet.pyx
+4 -4
View File
@@ -568,7 +568,7 @@ cdef class FileMetaData:
def __reduce__(self):
cdef:
NativeFile sink = BufferOutputStream()
OutputStream* c_sink = sink.get_output_stream().get()
COutputStream* c_sink = sink.get_output_stream().get()
with nogil:
self._metadata.WriteTo(c_sink)
@@ -694,7 +694,7 @@ cdef class FileMetaData:
Write the metadata object to a metadata-only file
"""
cdef:
shared_ptr[OutputStream] sink
shared_ptr[COutputStream] sink
c_string c_where
try:
@@ -1010,7 +1010,7 @@ cdef class ParquetReader:
read_dictionary=None, FileMetaData metadata=None,
int buffer_size=0):
cdef:
shared_ptr[RandomAccessFile] rd_handle
shared_ptr[CRandomAccessFile] rd_handle
shared_ptr[CFileMetaData] c_metadata
CReaderProperties properties = default_reader_properties()
ArrowReaderProperties arrow_props = (
@@ -1202,7 +1202,7 @@ cdef class ParquetReader:
cdef class ParquetWriter:
cdef:
unique_ptr[FileWriter] writer
shared_ptr[OutputStream] sink
shared_ptr[COutputStream] sink
bint own_sink
cdef readonly:
python/pyarrow/_s3.pyx
+52
View File
@@ -0,0 +1,52 @@
# Licensed to the Apache Software Foundation (ASF) under one
# or more contributor license agreements. See the NOTICE file
# distributed with this work for additional information
# regarding copyright ownership. The ASF licenses this file
# to you under the Apache License, Version 2.0 (the
# "License"); you may not use this file except in compliance
# with the License. You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing,
# software distributed under the License is distributed on an
# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
# KIND, either express or implied. See the License for the
# specific language governing permissions and limitations
# under the License.
# cython: language_level = 3
import six
from pyarrow.compat import frombytes, tobytes
from pyarrow.includes.common cimport *
from pyarrow.includes.libarrow cimport PyDateTime_from_TimePoint
from pyarrow.lib import _detect_compression
from pyarrow.lib cimport *
cdef class S3FileSystem(FileSystem):
cdef:
CS3FileSystem* s3fs
def __init__(self, str access_key, str secret_key, str region='us-east-1',
str scheme='https', str endpoint_override=None):
cdef:
CS3Options options
shared_ptr[CS3FileSystem] wrapped
options.access_key = tobytes(access_key)
options.secret_key = tobytes(secret_key)
options.region = tobytes(region)
options.scheme = tobytes(scheme)
if endpoint_override is not None:
options.endpoint_override = endpoint_override
check_status(CS3FileSystem.Make(options, &wrapped))
self.init(<shared_ptr[CFileSystem]> wrapped)
cdef init(self, const shared_ptr[CFileSystem]& wrapped):
FileSystem.init(self, wrapped)
self.s3fs = <CS3FileSystem*> wrapped.get()
python/pyarrow/feather.pxi
+2 -2
View File
@@ -34,7 +34,7 @@ cdef class FeatherWriter:
self.num_rows = -1
def open(self, object dest):
cdef shared_ptr[OutputStream] sink
cdef shared_ptr[COutputStream] sink
get_writer(dest, &sink)
with nogil:
@@ -76,7 +76,7 @@ cdef class FeatherReader:
pass
def open(self, source, c_bool use_memory_map=True):
cdef shared_ptr[RandomAccessFile] reader
cdef shared_ptr[CRandomAccessFile] reader
get_reader(source, use_memory_map, &reader)
with nogil:
python/pyarrow/fs.py
+5
View File
@@ -18,3 +18,8 @@
from __future__ import absolute_import
from pyarrow._fs import * # noqa
try:
from pyarrow._s3 import * # noqa
except ImportError:
pass
python/pyarrow/includes/libarrow.pxd
+117 -49
View File
@@ -765,13 +765,16 @@ cdef extern from "arrow/io/api.h" namespace "arrow::io" nogil:
CStatus Write(const uint8_t* data, int64_t nbytes)
CStatus Flush()
cdef cppclass OutputStream(FileInterface, Writable):
cdef cppclass COutputStream" arrow::io::OutputStream"(FileInterface,
Writable):
pass
cdef cppclass InputStream(FileInterface, Readable):
cdef cppclass CInputStream" arrow::io::InputStream"(FileInterface,
Readable):
pass
cdef cppclass RandomAccessFile(InputStream, Seekable):
cdef cppclass CRandomAccessFile" arrow::io::RandomAccessFile"(CInputStream,
Seekable):
CStatus GetSize(int64_t* size)
CStatus ReadAt(int64_t position, int64_t nbytes,
@@ -780,24 +783,24 @@ cdef extern from "arrow/io/api.h" namespace "arrow::io" nogil:
shared_ptr[CBuffer]* out)
c_bool supports_zero_copy()
cdef cppclass WritableFile(OutputStream, Seekable):
cdef cppclass WritableFile(COutputStream, Seekable):
CStatus WriteAt(int64_t position, const uint8_t* data,
int64_t nbytes)
cdef cppclass ReadWriteFileInterface(RandomAccessFile,
cdef cppclass ReadWriteFileInterface(CRandomAccessFile,
WritableFile):
pass
cdef cppclass FileSystem:
cdef cppclass CIOFileSystem" arrow::io::FileSystem":
CStatus Stat(const c_string& path, FileStatistics* stat)
cdef cppclass FileOutputStream(OutputStream):
cdef cppclass FileOutputStream(COutputStream):
@staticmethod
CStatus Open(const c_string& path, shared_ptr[OutputStream]* file)
CStatus Open(const c_string& path, shared_ptr[COutputStream]* file)
int file_descriptor()
cdef cppclass ReadableFile(RandomAccessFile):
cdef cppclass ReadableFile(CRandomAccessFile):
@staticmethod
CStatus Open(const c_string& path, shared_ptr[ReadableFile]* file)
@@ -823,46 +826,46 @@ cdef extern from "arrow/io/api.h" namespace "arrow::io" nogil:
int file_descriptor()
cdef cppclass CCompressedInputStream \
" arrow::io::CompressedInputStream"(InputStream):
" arrow::io::CompressedInputStream"(CInputStream):
@staticmethod
CStatus Make(CMemoryPool* pool, CCodec* codec,
shared_ptr[InputStream] raw,
shared_ptr[CInputStream] raw,
shared_ptr[CCompressedInputStream]* out)
@staticmethod
CStatus Make(CCodec* codec, shared_ptr[InputStream] raw,
CStatus Make(CCodec* codec, shared_ptr[CInputStream] raw,
shared_ptr[CCompressedInputStream]* out)
cdef cppclass CCompressedOutputStream \
" arrow::io::CompressedOutputStream"(OutputStream):
" arrow::io::CompressedOutputStream"(COutputStream):
@staticmethod
CStatus Make(CMemoryPool* pool, CCodec* codec,
shared_ptr[OutputStream] raw,
shared_ptr[COutputStream] raw,
shared_ptr[CCompressedOutputStream]* out)
@staticmethod
CStatus Make(CCodec* codec, shared_ptr[OutputStream] raw,
CStatus Make(CCodec* codec, shared_ptr[COutputStream] raw,
shared_ptr[CCompressedOutputStream]* out)
cdef cppclass CBufferedInputStream \
" arrow::io::BufferedInputStream"(InputStream):
" arrow::io::BufferedInputStream"(CInputStream):
@staticmethod
CStatus Create(int64_t buffer_size, CMemoryPool* pool,
shared_ptr[InputStream] raw,
shared_ptr[CInputStream] raw,
shared_ptr[CBufferedInputStream]* out)
shared_ptr[InputStream] Detach()
shared_ptr[CInputStream] Detach()
cdef cppclass CBufferedOutputStream \
" arrow::io::BufferedOutputStream"(OutputStream):
" arrow::io::BufferedOutputStream"(COutputStream):
@staticmethod
CStatus Create(int64_t buffer_size, CMemoryPool* pool,
shared_ptr[OutputStream] raw,
shared_ptr[COutputStream] raw,
shared_ptr[CBufferedOutputStream]* out)
CStatus Detach(shared_ptr[OutputStream]* raw)
CStatus Detach(shared_ptr[COutputStream]* raw)
# ----------------------------------------------------------------------
# HDFS
@@ -894,13 +897,13 @@ cdef extern from "arrow/io/api.h" namespace "arrow::io" nogil:
int64_t block_size
int16_t permissions
cdef cppclass HdfsReadableFile(RandomAccessFile):
cdef cppclass HdfsReadableFile(CRandomAccessFile):
pass
cdef cppclass HdfsOutputStream(OutputStream):
cdef cppclass HdfsOutputStream(COutputStream):
pass
cdef cppclass CHadoopFileSystem" arrow::io::HadoopFileSystem"(FileSystem):
cdef cppclass CHadoopFileSystem" arrow::io::HadoopFileSystem"(CIOFileSystem):
@staticmethod
CStatus Connect(const HdfsConnectionConfig* config,
shared_ptr[CHadoopFileSystem]* client)
@@ -936,16 +939,16 @@ cdef extern from "arrow/io/api.h" namespace "arrow::io" nogil:
shared_ptr[HdfsOutputStream]* handle)
cdef cppclass CBufferReader \
" arrow::io::BufferReader"(RandomAccessFile):
" arrow::io::BufferReader"(CRandomAccessFile):
CBufferReader(const shared_ptr[CBuffer]& buffer)
CBufferReader(const uint8_t* data, int64_t nbytes)
cdef cppclass CBufferOutputStream \
" arrow::io::BufferOutputStream"(OutputStream):
" arrow::io::BufferOutputStream"(COutputStream):
CBufferOutputStream(const shared_ptr[CResizableBuffer]& buffer)
cdef cppclass CMockOutputStream \
" arrow::io::MockOutputStream"(OutputStream):
" arrow::io::MockOutputStream"(COutputStream):
CMockOutputStream()
int64_t GetExtentBytesWritten()
@@ -958,6 +961,71 @@ cdef extern from "arrow/io/api.h" namespace "arrow::io" nogil:
void set_memcopy_threshold(int64_t threshold)
cdef extern from "arrow/filesystem/api.h" namespace "arrow::fs" nogil:
enum CFileType "arrow::fs::FileType":
CFileType_NonExistent "arrow::fs::FileType::NonExistent"
CFileType_Unknown "arrow::fs::FileType::Unknown"
CFileType_File "arrow::fs::FileType::File"
CFileType_Directory "arrow::fs::FileType::Directory"
cdef cppclass CTimePoint "arrow::fs::TimePoint":
pass
cdef cppclass CFileStats "arrow::fs::FileStats":
CFileStats()
CFileStats(CFileStats&&)
CFileStats& operator=(CFileStats&&)
CFileStats(const CFileStats&)
CFileStats& operator=(const CFileStats&)
CFileType type()
void set_type(CFileType type)
c_string path()
void set_path(const c_string& path)
c_string base_name()
int64_t size()
void set_size(int64_t size)
c_string extension()
CTimePoint mtime()
void set_mtime(CTimePoint mtime)
cdef cppclass CSelector "arrow::fs::Selector":
CSelector()
c_string base_dir
c_bool allow_non_existent
c_bool recursive
cdef cppclass CFileSystem "arrow::fs::FileSystem":
CStatus GetTargetStats(const c_string& path, CFileStats* out)
CStatus GetTargetStats(const vector[c_string]& paths,
vector[CFileStats]* out)
CStatus GetTargetStats(const CSelector& select,
vector[CFileStats]* out)
CStatus CreateDir(const c_string& path, c_bool recursive)
CStatus DeleteDir(const c_string& path)
CStatus DeleteFile(const c_string& path)
CStatus DeleteFiles(const vector[c_string]& paths)
CStatus Move(const c_string& src, const c_string& dest)
CStatus CopyFile(const c_string& src, const c_string& dest)
CStatus OpenInputStream(const c_string& path,
shared_ptr[CInputStream]* out)
CStatus OpenInputFile(const c_string& path,
shared_ptr[CRandomAccessFile]* out)
CStatus OpenOutputStream(const c_string& path,
shared_ptr[COutputStream]* out)
CStatus OpenAppendStream(const c_string& path,
shared_ptr[COutputStream]* out)
cdef cppclass CLocalFileSystem "arrow::fs::LocalFileSystem"(CFileSystem):
LocalFileSystem()
cdef cppclass CSubTreeFileSystem \
"arrow::fs::SubTreeFileSystem"(CFileSystem):
CSubTreeFileSystem(const c_string& base_path,
shared_ptr[CFileSystem] base_fs)
cdef extern from "arrow/ipc/api.h" namespace "arrow::ipc" nogil:
enum MessageType" arrow::ipc::Message::Type":
MessageType_SCHEMA" arrow::ipc::Message::SCHEMA"
@@ -1001,14 +1069,14 @@ cdef extern from "arrow/ipc/api.h" namespace "arrow::ipc" nogil:
MetadataVersion metadata_version()
MessageType type()
CStatus SerializeTo(OutputStream* stream, const CIpcOptions& options,
CStatus SerializeTo(COutputStream* stream, const CIpcOptions& options,
int64_t* output_length)
c_string FormatMessageType(MessageType type)
cdef cppclass CMessageReader" arrow::ipc::MessageReader":
@staticmethod
unique_ptr[CMessageReader] Open(const shared_ptr[InputStream]& stream)
unique_ptr[CMessageReader] Open(const shared_ptr[CInputStream]& stream)
CStatus ReadNextMessage(unique_ptr[CMessage]* out)
@@ -1020,7 +1088,7 @@ cdef extern from "arrow/ipc/api.h" namespace "arrow::ipc" nogil:
cdef cppclass CRecordBatchStreamReader \
" arrow::ipc::RecordBatchStreamReader"(CRecordBatchReader):
@staticmethod
CStatus Open(const InputStream* stream,
CStatus Open(const CInputStream* stream,
shared_ptr[CRecordBatchReader]* out)
@staticmethod
@@ -1031,24 +1099,24 @@ cdef extern from "arrow/ipc/api.h" namespace "arrow::ipc" nogil:
" arrow::ipc::RecordBatchStreamWriter"(CRecordBatchWriter):
@staticmethod
CResult[shared_ptr[CRecordBatchWriter]] Open(
OutputStream* sink, const shared_ptr[CSchema]& schema,
COutputStream* sink, const shared_ptr[CSchema]& schema,
CIpcOptions& options)
cdef cppclass CRecordBatchFileWriter \
" arrow::ipc::RecordBatchFileWriter"(CRecordBatchWriter):
@staticmethod
CResult[shared_ptr[CRecordBatchWriter]] Open(
OutputStream* sink, const shared_ptr[CSchema]& schema,
COutputStream* sink, const shared_ptr[CSchema]& schema,
CIpcOptions& options)
cdef cppclass CRecordBatchFileReader \
" arrow::ipc::RecordBatchFileReader":
@staticmethod
CStatus Open(RandomAccessFile* file,
CStatus Open(CRandomAccessFile* file,
shared_ptr[CRecordBatchFileReader]* out)
@staticmethod
CStatus Open2" Open"(RandomAccessFile* file,
CStatus Open2" Open"(CRandomAccessFile* file,
int64_t footer_offset,
shared_ptr[CRecordBatchFileReader]* out)
@@ -1058,16 +1126,16 @@ cdef extern from "arrow/ipc/api.h" namespace "arrow::ipc" nogil:
CStatus ReadRecordBatch(int i, shared_ptr[CRecordBatch]* batch)
CStatus ReadMessage(InputStream* stream, unique_ptr[CMessage]* message)
CStatus ReadMessage(CInputStream* stream, unique_ptr[CMessage]* message)
CStatus GetRecordBatchSize(const CRecordBatch& batch, int64_t* size)
CStatus GetTensorSize(const CTensor& tensor, int64_t* size)
CStatus WriteTensor(const CTensor& tensor, OutputStream* dst,
CStatus WriteTensor(const CTensor& tensor, COutputStream* dst,
int32_t* metadata_length,
int64_t* body_length)
CStatus ReadTensor(InputStream* stream, shared_ptr[CTensor]* out)
CStatus ReadTensor(CInputStream* stream, shared_ptr[CTensor]* out)
CStatus ReadRecordBatch(const CMessage& message,
const shared_ptr[CSchema]& schema,
@@ -1082,16 +1150,16 @@ cdef extern from "arrow/ipc/api.h" namespace "arrow::ipc" nogil:
CMemoryPool* pool,
shared_ptr[CBuffer]* out)
CStatus ReadSchema(InputStream* stream, CDictionaryMemo* dictionary_memo,
CStatus ReadSchema(CInputStream* stream, CDictionaryMemo* dictionary_memo,
shared_ptr[CSchema]* out)
CStatus ReadRecordBatch(const shared_ptr[CSchema]& schema,
CDictionaryMemo* dictionary_memo,
InputStream* stream,
CInputStream* stream,
shared_ptr[CRecordBatch]* out)
CStatus AlignStream(InputStream* stream, int64_t alignment)
CStatus AlignStream(OutputStream* stream, int64_t alignment)
CStatus AlignStream(CInputStream* stream, int64_t alignment)
CStatus AlignStream(COutputStream* stream, int64_t alignment)
cdef CStatus GetRecordBatchPayload\
" arrow::ipc::internal::GetRecordBatchPayload"(
@@ -1102,7 +1170,7 @@ cdef extern from "arrow/ipc/api.h" namespace "arrow::ipc" nogil:
cdef cppclass CFeatherWriter" arrow::ipc::feather::TableWriter":
@staticmethod
CStatus Open(const shared_ptr[OutputStream]& stream,
CStatus Open(const shared_ptr[COutputStream]& stream,
unique_ptr[CFeatherWriter]* out)
void SetDescription(const c_string& desc)
@@ -1113,7 +1181,7 @@ cdef extern from "arrow/ipc/api.h" namespace "arrow::ipc" nogil:
cdef cppclass CFeatherReader" arrow::ipc::feather::TableReader":
@staticmethod
CStatus Open(const shared_ptr[RandomAccessFile]& file,
CStatus Open(const shared_ptr[CRandomAccessFile]& file,
unique_ptr[CFeatherReader]* out)
c_string GetDescription()
@@ -1172,7 +1240,7 @@ cdef extern from "arrow/csv/api.h" namespace "arrow::csv" nogil:
cdef cppclass CCSVReader" arrow::csv::TableReader":
@staticmethod
CStatus Make(CMemoryPool*, shared_ptr[InputStream],
CStatus Make(CMemoryPool*, shared_ptr[CInputStream],
CCSVReadOptions, CCSVParseOptions, CCSVConvertOptions,
shared_ptr[CCSVReader]* out)
@@ -1200,7 +1268,7 @@ cdef extern from "arrow/json/reader.h" namespace "arrow::json" nogil:
cdef cppclass CJSONReader" arrow::json::TableReader":
@staticmethod
CStatus Make(CMemoryPool*, shared_ptr[InputStream],
CStatus Make(CMemoryPool*, shared_ptr[CInputStream],
CJSONReadOptions, CJSONParseOptions,
shared_ptr[CJSONReader]* out)
@@ -1379,10 +1447,10 @@ cdef extern from "arrow/python/api.h" namespace "arrow::py" nogil:
CStatus Make(const uint8_t* data, int64_t size, object base,
shared_ptr[CBuffer]* out)
cdef cppclass PyReadableFile(RandomAccessFile):
cdef cppclass PyReadableFile(CRandomAccessFile):
PyReadableFile(object fo)
cdef cppclass PyOutputStream(OutputStream):
cdef cppclass PyOutputStream(COutputStream):
PyOutputStream(object fo)
cdef cppclass PandasOptions:
@@ -1398,7 +1466,7 @@ cdef extern from "arrow/python/api.h" namespace "arrow::py" nogil:
shared_ptr[CRecordBatch] batch
vector[shared_ptr[CTensor]] tensors
CStatus WriteTo(OutputStream* dst)
CStatus WriteTo(COutputStream* dst)
CStatus GetComponents(CMemoryPool* pool, PyObject** dst)
CStatus SerializeObject(object context, object sequence,
@@ -1408,7 +1476,7 @@ cdef extern from "arrow/python/api.h" namespace "arrow::py" nogil:
const CSerializedPyObject& obj,
PyObject* base, PyObject** out)
CStatus ReadSerializedObject(RandomAccessFile* src,
CStatus ReadSerializedObject(CRandomAccessFile* src,
CSerializedPyObject* out)
CStatus GetSerializedFromComponents(int num_tensors, int num_ndarrays,
python/pyarrow/includes/libarrow_fs.pxd
-92
View File
@@ -1,92 +0,0 @@
# Licensed to the Apache Software Foundation (ASF) under one
# or more contributor license agreements. See the NOTICE file
# distributed with this work for additional information
# regarding copyright ownership. The ASF licenses this file
# to you under the Apache License, Version 2.0 (the
# "License"); you may not use this file except in compliance
# with the License. You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing,
# software distributed under the License is distributed on an
# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
# KIND, either express or implied. See the License for the
# specific language governing permissions and limitations
# under the License.
# distutils: language = c++
from libcpp.functional cimport function
from pyarrow.includes.common cimport *
from pyarrow.includes.libarrow cimport (
InputStream as CInputStream,
OutputStream as COutputStream,
RandomAccessFile as CRandomAccessFile
)
cdef extern from "arrow/filesystem/api.h" namespace "arrow::fs" nogil:
enum CFileType "arrow::fs::FileType":
CFileType_NonExistent "arrow::fs::FileType::NonExistent"
CFileType_Unknown "arrow::fs::FileType::Unknown"
CFileType_File "arrow::fs::FileType::File"
CFileType_Directory "arrow::fs::FileType::Directory"
cdef cppclass CTimePoint "arrow::fs::TimePoint":
pass
cdef cppclass CFileStats "arrow::fs::FileStats":
CFileStats()
CFileStats(CFileStats&&)
CFileStats& operator=(CFileStats&&)
CFileStats(const CFileStats&)
CFileStats& operator=(const CFileStats&)
CFileType type()
void set_type(CFileType type)
c_string path()
void set_path(const c_string& path)
c_string base_name()
int64_t size()
void set_size(int64_t size)
c_string extension()
CTimePoint mtime()
void set_mtime(CTimePoint mtime)
cdef cppclass CSelector "arrow::fs::Selector":
CSelector()
c_string base_dir
c_bool allow_non_existent
c_bool recursive
cdef cppclass CFileSystem "arrow::fs::FileSystem":
CStatus GetTargetStats(const c_string& path, CFileStats* out)
CStatus GetTargetStats(const vector[c_string]& paths,
vector[CFileStats]* out)
CStatus GetTargetStats(const CSelector& select,
vector[CFileStats]* out)
CStatus CreateDir(const c_string& path, c_bool recursive)
CStatus DeleteDir(const c_string& path)
CStatus DeleteFile(const c_string& path)
CStatus DeleteFiles(const vector[c_string]& paths)
CStatus Move(const c_string& src, const c_string& dest)
CStatus CopyFile(const c_string& src, const c_string& dest)
CStatus OpenInputStream(const c_string& path,
shared_ptr[CInputStream]* out)
CStatus OpenInputFile(const c_string& path,
shared_ptr[CRandomAccessFile]* out)
CStatus OpenOutputStream(const c_string& path,
shared_ptr[COutputStream]* out)
CStatus OpenAppendStream(const c_string& path,
shared_ptr[COutputStream]* out)
cdef cppclass CLocalFileSystem "arrow::fs::LocalFileSystem"(CFileSystem):
LocalFileSystem()
cdef cppclass CSubTreeFileSystem \
"arrow::fs::SubTreeFileSystem"(CFileSystem):
CSubTreeFileSystem(const c_string& base_path,
shared_ptr[CFileSystem] base_fs)
python/pyarrow/includes/libarrow_s3.pxd
+36
View File
@@ -0,0 +1,36 @@
# Licensed to the Apache Software Foundation (ASF) under one
# or more contributor license agreements. See the NOTICE file
# distributed with this work for additional information
# regarding copyright ownership. The ASF licenses this file
# to you under the Apache License, Version 2.0 (the
# "License"); you may not use this file except in compliance
# with the License. You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing,
# software distributed under the License is distributed on an
# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
# KIND, either express or implied. See the License for the
# specific language governing permissions and limitations
# under the License.
# distutils: language = c++
from libcpp.functional cimport function
from pyarrow.includes.common cimport *
from pyarrow.includes.libarrow cimport CFileSystem
cdef extern from "arrow/filesystem/api.h" namespace "arrow::fs" nogil:
cdef struct CS3Options "arrow::fs::S3Options":
c_string region
c_string endpoint_override
c_string scheme
c_string access_key
c_string secret_key
cdef cppclass CS3FileSystem "arrow::fs::S3FileSystem"(CFileSystem):
@staticmethod
CStatus Make(const CS3Options& options, shared_ptr[CS3FileSystem]* out)
python/pyarrow/io-hdfs.pxi
+2 -2
View File
@@ -424,7 +424,7 @@ cdef class HadoopFileSystem:
c_replication, c_default_block_size,
&wr_handle))
out.set_output_stream(<shared_ptr[OutputStream]> wr_handle)
out.set_output_stream(<shared_ptr[COutputStream]> wr_handle)
out.is_writable = True
else:
with nogil:
@@ -432,7 +432,7 @@ cdef class HadoopFileSystem:
.OpenReadable(c_path, &rd_handle))
out.set_random_access_file(
<shared_ptr[RandomAccessFile]> rd_handle)
<shared_ptr[CRandomAccessFile]> rd_handle)
out.is_readable = True
assert not out.closed
python/pyarrow/io.pxi
+28 -28
View File
@@ -133,29 +133,29 @@ cdef class NativeFile:
else:
check_status(self.output_stream.get().Close())
cdef set_random_access_file(self, shared_ptr[RandomAccessFile] handle):
self.input_stream = <shared_ptr[InputStream]> handle
cdef set_random_access_file(self, shared_ptr[CRandomAccessFile] handle):
self.input_stream = <shared_ptr[CInputStream]> handle
self.random_access = handle
self.is_seekable = True
cdef set_input_stream(self, shared_ptr[InputStream] handle):
cdef set_input_stream(self, shared_ptr[CInputStream] handle):
self.input_stream = handle
self.random_access.reset()
self.is_seekable = False
cdef set_output_stream(self, shared_ptr[OutputStream] handle):
cdef set_output_stream(self, shared_ptr[COutputStream] handle):
self.output_stream = handle
cdef shared_ptr[RandomAccessFile] get_random_access_file(self) except *:
cdef shared_ptr[CRandomAccessFile] get_random_access_file(self) except *:
self._assert_readable()
self._assert_seekable()
return self.random_access
cdef shared_ptr[InputStream] get_input_stream(self) except *:
cdef shared_ptr[CInputStream] get_input_stream(self) except *:
self._assert_readable()
return self.input_stream
cdef shared_ptr[OutputStream] get_output_stream(self) except *:
cdef shared_ptr[COutputStream] get_output_stream(self) except *:
self._assert_writable()
return self.output_stream
@@ -679,11 +679,11 @@ cdef class PythonFile(NativeFile):
if kind == 'r':
self.set_random_access_file(
shared_ptr[RandomAccessFile](new PyReadableFile(handle)))
shared_ptr[CRandomAccessFile](new PyReadableFile(handle)))
self.is_readable = True
else:
self.set_output_stream(
shared_ptr[OutputStream](new PyOutputStream(handle)))
shared_ptr[COutputStream](new PyOutputStream(handle)))
self.is_writable = True
def truncate(self, pos=None):
@@ -720,8 +720,8 @@ cdef class MemoryMappedFile(NativeFile):
result.path = path
result.is_readable = True
result.is_writable = True
result.set_output_stream(<shared_ptr[OutputStream]> handle)
result.set_random_access_file(<shared_ptr[RandomAccessFile]> handle)
result.set_output_stream(<shared_ptr[COutputStream]> handle)
result.set_random_access_file(<shared_ptr[CRandomAccessFile]> handle)
result.handle = handle
return result
@@ -750,8 +750,8 @@ cdef class MemoryMappedFile(NativeFile):
with nogil:
check_status(CMemoryMappedFile.Open(c_path, c_mode, &handle))
self.set_output_stream(<shared_ptr[OutputStream]> handle)
self.set_random_access_file(<shared_ptr[RandomAccessFile]> handle)
self.set_output_stream(<shared_ptr[COutputStream]> handle)
self.set_random_access_file(<shared_ptr[CRandomAccessFile]> handle)
self.handle = handle
def resize(self, new_size):
@@ -836,7 +836,7 @@ cdef class OSFile(NativeFile):
check_status(ReadableFile.Open(path, pool, &handle))
self.is_readable = True
self.set_random_access_file(<shared_ptr[RandomAccessFile]> handle)
self.set_random_access_file(<shared_ptr[CRandomAccessFile]> handle)
cdef _open_writable(self, c_string path):
with nogil:
@@ -1174,7 +1174,7 @@ cdef class BufferReader(NativeFile):
def __cinit__(self, object obj):
self.buffer = as_buffer(obj)
self.set_random_access_file(shared_ptr[RandomAccessFile](
self.set_random_access_file(shared_ptr[CRandomAccessFile](
new CBufferReader(self.buffer.buffer)))
self.is_readable = True
@@ -1204,7 +1204,7 @@ cdef class CompressedInputStream(NativeFile):
check_status(CCompressedInputStream.Make(
codec.get(), stream.get_input_stream(), &compressed_stream))
self.set_input_stream(<shared_ptr[InputStream]> compressed_stream)
self.set_input_stream(<shared_ptr[CInputStream]> compressed_stream)
self.is_readable = True
@@ -1234,13 +1234,13 @@ cdef class CompressedOutputStream(NativeFile):
check_status(CCompressedOutputStream.Make(
codec.get(), stream.get_output_stream(), &compressed_stream))
self.set_output_stream(<shared_ptr[OutputStream]> compressed_stream)
self.set_output_stream(<shared_ptr[COutputStream]> compressed_stream)
self.is_writable = True
ctypedef CBufferedInputStream* _CBufferedInputStreamPtr
ctypedef CBufferedOutputStream* _CBufferedOutputStreamPtr
ctypedef RandomAccessFile* _RandomAccessFilePtr
ctypedef CRandomAccessFile* _RandomAccessFilePtr
cdef class BufferedInputStream(NativeFile):
@@ -1255,7 +1255,7 @@ cdef class BufferedInputStream(NativeFile):
buffer_size, maybe_unbox_memory_pool(memory_pool),
stream.get_input_stream(), &buffered_stream))
self.set_input_stream(<shared_ptr[InputStream]> buffered_stream)
self.set_input_stream(<shared_ptr[CInputStream]> buffered_stream)
self.is_readable = True
def detach(self):
@@ -1269,7 +1269,7 @@ cdef class BufferedInputStream(NativeFile):
The underlying raw input stream
"""
cdef:
shared_ptr[InputStream] c_raw
shared_ptr[CInputStream] c_raw
_CBufferedInputStreamPtr buffered
NativeFile raw
@@ -1287,7 +1287,7 @@ cdef class BufferedInputStream(NativeFile):
# selectively.
if dynamic_cast[_RandomAccessFilePtr](c_raw.get()) != nullptr:
raw.set_random_access_file(
static_pointer_cast[RandomAccessFile, InputStream](c_raw))
static_pointer_cast[CRandomAccessFile, CInputStream](c_raw))
else:
raw.set_input_stream(c_raw)
return raw
@@ -1305,7 +1305,7 @@ cdef class BufferedOutputStream(NativeFile):
buffer_size, maybe_unbox_memory_pool(memory_pool),
stream.get_output_stream(), &buffered_stream))
self.set_output_stream(<shared_ptr[OutputStream]> buffered_stream)
self.set_output_stream(<shared_ptr[COutputStream]> buffered_stream)
self.is_writable = True
def detach(self):
@@ -1319,7 +1319,7 @@ cdef class BufferedOutputStream(NativeFile):
The underlying raw output stream
"""
cdef:
shared_ptr[OutputStream] c_raw
shared_ptr[COutputStream] c_raw
_CBufferedOutputStreamPtr buffered
NativeFile raw
@@ -1400,7 +1400,7 @@ cdef NativeFile _get_native_file(object source, c_bool use_memory_map):
cdef get_reader(object source, c_bool use_memory_map,
shared_ptr[RandomAccessFile]* reader):
shared_ptr[CRandomAccessFile]* reader):
cdef NativeFile nf
nf = _get_native_file(source, use_memory_map)
@@ -1408,7 +1408,7 @@ cdef get_reader(object source, c_bool use_memory_map,
cdef get_input_stream(object source, c_bool use_memory_map,
shared_ptr[InputStream]* out):
shared_ptr[CInputStream]* out):
"""
Like get_reader(), but can automatically decompress, and returns
an InputStream.
@@ -1416,7 +1416,7 @@ cdef get_input_stream(object source, c_bool use_memory_map,
cdef:
NativeFile nf
unique_ptr[CCodec] codec
shared_ptr[InputStream] input_stream
shared_ptr[CInputStream] input_stream
shared_ptr[CCompressedInputStream] compressed_stream
CompressionType compression_type
@@ -1435,12 +1435,12 @@ cdef get_input_stream(object source, c_bool use_memory_map,
check_status(CCodec.Create(compression_type, &codec))
check_status(CCompressedInputStream.Make(codec.get(), input_stream,
&compressed_stream))
input_stream = <shared_ptr[InputStream]> compressed_stream
input_stream = <shared_ptr[CInputStream]> compressed_stream
out[0] = input_stream
cdef get_writer(object source, shared_ptr[OutputStream]* writer):
cdef get_writer(object source, shared_ptr[COutputStream]* writer):
cdef NativeFile nf
try:
python/pyarrow/ipc.pxi
+15 -13
View File
@@ -17,6 +17,7 @@
import warnings
cdef class Message:
"""
Container for an Arrow IPC message with metadata and optional body
@@ -76,7 +77,7 @@ cdef class Message:
"""
cdef:
int64_t output_length = 0
OutputStream* out
COutputStream* out
CIpcOptions options
options.alignment = alignment
@@ -136,9 +137,11 @@ cdef class MessageReader:
@staticmethod
def open_stream(source):
cdef MessageReader result = MessageReader.__new__(MessageReader)
cdef shared_ptr[InputStream] in_stream
cdef unique_ptr[CMessageReader] reader
cdef:
MessageReader result = MessageReader.__new__(MessageReader)
shared_ptr[CInputStream] in_stream
unique_ptr[CMessageReader] reader
_get_input_stream(source, &in_stream)
with nogil:
reader = CMessageReader.Open(in_stream)
@@ -250,7 +253,7 @@ cdef class _CRecordBatchWriter:
cdef class _RecordBatchStreamWriter(_CRecordBatchWriter):
cdef:
shared_ptr[OutputStream] sink
shared_ptr[COutputStream] sink
CIpcOptions options
bint closed
@@ -276,7 +279,7 @@ cdef class _RecordBatchStreamWriter(_CRecordBatchWriter):
self.writer = GetResultValue(result)
cdef _get_input_stream(object source, shared_ptr[InputStream]* out):
cdef _get_input_stream(object source, shared_ptr[CInputStream]* out):
try:
source = as_buffer(source)
except TypeError:
@@ -332,7 +335,7 @@ cdef class _CRecordBatchReader:
cdef class _RecordBatchStreamReader(_CRecordBatchReader):
cdef:
shared_ptr[InputStream] in_stream
shared_ptr[CInputStream] in_stream
cdef readonly:
Schema schema
@@ -367,7 +370,7 @@ cdef class _RecordBatchFileWriter(_RecordBatchStreamWriter):
cdef class _RecordBatchFileReader:
cdef:
shared_ptr[CRecordBatchFileReader] reader
shared_ptr[RandomAccessFile] file
shared_ptr[CRandomAccessFile] file
cdef readonly:
Schema schema
@@ -516,9 +519,8 @@ def read_tensor(source):
"""
cdef:
shared_ptr[CTensor] sp_tensor
InputStream* c_stream
cdef NativeFile nf = as_native_file(source)
CInputStream* c_stream
NativeFile nf = as_native_file(source)
c_stream = nf.get_input_stream().get()
with nogil:
@@ -540,7 +542,7 @@ def read_message(source):
"""
cdef:
Message result = Message.__new__(Message)
InputStream* c_stream
CInputStream* c_stream
cdef NativeFile nf = as_native_file(source)
c_stream = nf.get_input_stream().get()
@@ -571,7 +573,7 @@ def read_schema(obj, DictionaryMemo dictionary_memo=None):
"""
cdef:
shared_ptr[CSchema] result
shared_ptr[RandomAccessFile] cpp_file
shared_ptr[CRandomAccessFile] cpp_file
CDictionaryMemo temp_memo
CDictionaryMemo* arg_dict_memo
python/pyarrow/lib.pxd
+12 -12
View File
@@ -436,9 +436,9 @@ cdef class ResizableBuffer(Buffer):
cdef class NativeFile:
cdef:
shared_ptr[InputStream] input_stream
shared_ptr[RandomAccessFile] random_access
shared_ptr[OutputStream] output_stream
shared_ptr[CInputStream] input_stream
shared_ptr[CRandomAccessFile] random_access
shared_ptr[COutputStream] output_stream
bint is_readable
bint is_writable
bint is_seekable
@@ -449,13 +449,13 @@ cdef class NativeFile:
# extension classes are technically virtual in the C++ sense) we can expose
# the arrow::io abstract file interfaces to other components throughout the
# suite of Arrow C++ libraries
cdef set_random_access_file(self, shared_ptr[RandomAccessFile] handle)
cdef set_input_stream(self, shared_ptr[InputStream] handle)
cdef set_output_stream(self, shared_ptr[OutputStream] handle)
cdef set_random_access_file(self, shared_ptr[CRandomAccessFile] handle)
cdef set_input_stream(self, shared_ptr[CInputStream] handle)
cdef set_output_stream(self, shared_ptr[COutputStream] handle)
cdef shared_ptr[RandomAccessFile] get_random_access_file(self) except *
cdef shared_ptr[InputStream] get_input_stream(self) except *
cdef shared_ptr[OutputStream] get_output_stream(self) except *
cdef shared_ptr[CRandomAccessFile] get_random_access_file(self) except *
cdef shared_ptr[CInputStream] get_input_stream(self) except *
cdef shared_ptr[COutputStream] get_output_stream(self) except *
cdef class BufferedInputStream(NativeFile):
@@ -485,10 +485,10 @@ cdef class _CRecordBatchReader:
cdef get_input_stream(object source, c_bool use_memory_map,
shared_ptr[InputStream]* reader)
shared_ptr[CInputStream]* reader)
cdef get_reader(object source, c_bool use_memory_map,
shared_ptr[RandomAccessFile]* reader)
cdef get_writer(object source, shared_ptr[OutputStream]* writer)
shared_ptr[CRandomAccessFile]* reader)
cdef get_writer(object source, shared_ptr[COutputStream]* writer)
# Default is allow_none=False
cdef DataType ensure_type(object type, c_bool allow_none=*)
python/pyarrow/serialization.pxi
+3 -3
View File
@@ -254,11 +254,11 @@ cdef class SerializedPyObject:
"""
Write serialized object to a sink
"""
cdef shared_ptr[OutputStream] stream
cdef shared_ptr[COutputStream] stream
get_writer(sink, &stream)
self._write_to(stream.get())
cdef _write_to(self, OutputStream* stream):
cdef _write_to(self, COutputStream* stream):
with nogil:
check_status(self.data.WriteTo(stream))
@@ -399,7 +399,7 @@ def read_serialized(source, base=None):
-------
serialized : the serialized data
"""
cdef shared_ptr[RandomAccessFile] stream
cdef shared_ptr[CRandomAccessFile] stream
get_reader(source, True, &stream)
cdef SerializedPyObject serialized = SerializedPyObject()
python/pyarrow/tests/conftest.py
+6
View File
@@ -126,6 +126,12 @@ try:
except ImportError:
pass
try:
from pyarrow.fs import S3FileSystem
defaults['s3'] = True
except ImportError:
pass
def pytest_configure(config):
for mark in groups:
python/pyarrow/tests/test_fs.py
+31
View File
@@ -15,6 +15,7 @@
# specific language governing permissions and limitations
# under the License.
import os
from datetime import datetime
try:
import pathlib
@@ -307,3 +308,33 @@ def test_open_append_stream(fs, tempdir, testpath, compression, buffer_size,
f.write(b'\nnewly added')
assert decompressor(file_.read_bytes()) == b'already existing\nnewly added'
import subprocess
@pytest.fixture
def minio_server(tempdir):
host, port = '127.0.0.1', 9000
access_key, secret_key = 'arrow', 'apachearrow'
datadir = tempdir / 'minio'
address = '{}:{}'.format(host, port)
args = ['minio', '--compat', 'server', '--address', address, str(datadir)]
env = os.environ.copy()
env.update({
'MINIO_ACCESS_KEY': access_key,
'MINIO_SECRET_KEY': secret_key
})
try:
with subprocess.Popen(args, env=env) as proc:
yield host, port
proc.terminate()
except FileNotFoundError as e:
pytest.skip('Minio executable cannot be located')
def test_minio(minio_server):
host, port = minio_server
python/setup.py
+6
View File
@@ -141,6 +141,8 @@ class build_ext(_build_ext):
if not hasattr(sys, 'gettotalrefcount'):
self.build_type = 'release'
self.with_s3 = strtobool(
os.environ.get('PYARROW_WITH_S3', '0'))
self.with_cuda = strtobool(
os.environ.get('PYARROW_WITH_CUDA', '0'))
self.with_flight = strtobool(
@@ -215,6 +217,8 @@ class build_ext(_build_ext):
if self.cmake_generator:
cmake_options += ['-G', self.cmake_generator]
if self.with_s3:
cmake_options.append('-DPYARROW_BUILD_S3=on')
if self.with_cuda:
cmake_options.append('-DPYARROW_BUILD_CUDA=on')
if self.with_flight:
@@ -414,6 +418,8 @@ class build_ext(_build_ext):
return True
if name == '_flight' and not self.with_flight:
return True
if name == '_s3' and not self.with_s3:
return True
if name == '_cuda' and not self.with_cuda:
return True
if name == 'gandiva' and not self.with_gandiva: