diff --git a/.travis.yml b/.travis.yml index 4fc143a493db..990fd2256674 100644 --- a/.travis.yml +++ b/.travis.yml @@ -87,6 +87,7 @@ matrix: - if [ $ARROW_CI_CPP_AFFECTED != "1" ] && [ $ARROW_CI_JAVA_AFFECTED != "1" ]; then exit; fi - $TRAVIS_BUILD_DIR/ci/travis_install_clang_tools.sh - $TRAVIS_BUILD_DIR/ci/travis_install_linux.sh + - $TRAVIS_BUILD_DIR/ci/travis_install_minio.sh # If either C++ or Python changed, we must install the C++ libraries - git submodule update --init - $TRAVIS_BUILD_DIR/ci/travis_before_script_cpp.sh @@ -110,12 +111,14 @@ matrix: - ARROW_TRAVIS_USE_SYSTEM_JAVA=1 - ARROW_TRAVIS_USE_TOOLCHAIN=1 - ARROW_TRAVIS_VALGRIND=1 + - ARROW_TRAVIS_S3=1 # TODO(wesm): Run the benchmarks outside of Travis # - ARROW_TRAVIS_PYTHON_BENCHMARKS=1 before_script: - if [ $ARROW_CI_PYTHON_AFFECTED != "1" ] && [ $ARROW_CI_DOCS_AFFECTED != "1" ]; then exit; fi - $TRAVIS_BUILD_DIR/ci/travis_install_clang_tools.sh - $TRAVIS_BUILD_DIR/ci/travis_install_linux.sh + - $TRAVIS_BUILD_DIR/ci/travis_install_minio.sh - $TRAVIS_BUILD_DIR/ci/travis_install_toolchain.sh script: - $TRAVIS_BUILD_DIR/ci/travis_script_java.sh || travis_terminate 1 @@ -136,6 +139,7 @@ matrix: - ARROW_TRAVIS_PLASMA=1 - ARROW_TRAVIS_FLIGHT=1 - ARROW_TRAVIS_ORC=1 + - ARROW_TRAVIS_S3=1 - ARROW_TRAVIS_PARQUET=1 # TODO(ARROW-4763): llvm and llvmdev packages are in conflict: # https://github.com/conda-forge/llvmdev-feedstock/issues/60 @@ -149,6 +153,7 @@ matrix: - if [ $ARROW_CI_CPP_AFFECTED != "1" ] && [ $ARROW_CI_JAVA_AFFECTED != "1" ]; then exit; fi # If either C++ or Python changed, we must install the C++ libraries - git submodule update --init + - $TRAVIS_BUILD_DIR/ci/travis_install_minio.sh - $TRAVIS_BUILD_DIR/ci/travis_before_script_cpp.sh script: - $TRAVIS_BUILD_DIR/ci/travis_script_cpp.sh || travis_terminate 1 @@ -161,6 +166,7 @@ matrix: cache: addons: env: + - ARROW_TRAVIS_S3=1 - ARROW_TRAVIS_PLASMA=1 - ARROW_TRAVIS_USE_TOOLCHAIN=1 - ARROW_BUILD_WARNING_LEVEL=CHECKIN @@ -170,6 +176,7 @@ matrix: before_script: script: - if [ $ARROW_CI_PYTHON_AFFECTED != "1" ]; then exit; fi + - $TRAVIS_BUILD_DIR/ci/travis_install_minio.sh - $TRAVIS_BUILD_DIR/ci/travis_script_python.sh 3.6 - name: "Java OpenJDK8 and OpenJDK11" language: cpp diff --git a/ci/conda_env_python.yml b/ci/conda_env_python.yml index a0cd737b326c..0e6e5bf8d5c3 100644 --- a/ci/conda_env_python.yml +++ b/ci/conda_env_python.yml @@ -22,6 +22,7 @@ numpy>=1.14 pandas pytest pytest-faulthandler +pytest-lazy-fixture pytz setuptools setuptools_scm=3.2.0 diff --git a/ci/cpp-msvc-build-main.bat b/ci/cpp-msvc-build-main.bat index b088e2eec763..b6d1b20a5109 100644 --- a/ci/cpp-msvc-build-main.bat +++ b/ci/cpp-msvc-build-main.bat @@ -98,6 +98,9 @@ pip install -r requirements.txt pickle5 set PYARROW_CXXFLAGS=%ARROW_CXXFLAGS% set PYARROW_CMAKE_GENERATOR=%GENERATOR% +if "%ARROW_S3%" == "ON" ( + set PYARROW_WITH_S3=ON +) if "%ARROW_BUILD_FLIGHT%" == "ON" ( @rem ARROW-5441: bundling Arrow Flight libraries not implemented set PYARROW_BUNDLE_ARROW_CPP=OFF diff --git a/ci/travis_install_linux.sh b/ci/travis_install_linux.sh index a5283139a833..441d50d9b16c 100755 --- a/ci/travis_install_linux.sh +++ b/ci/travis_install_linux.sh @@ -42,14 +42,6 @@ if [ "$ARROW_TRAVIS_GANDIVA" == "1" ]; then sudo apt-get install -y -qq llvm-$ARROW_LLVM_MAJOR_VERSION-dev fi -if [ "$ARROW_TRAVIS_S3" == "1" ]; then - # Download the Minio S3 server into PATH - S3FS_DIR=~/.local/bin/ - mkdir -p $S3FS_DIR - wget --directory-prefix $S3FS_DIR https://dl.min.io/server/minio/release/linux-amd64/minio - chmod +x $S3FS_DIR/minio -fi - if [ "$ARROW_TRAVIS_USE_SYSTEM" == "1" ]; then if [ "$DISTRO_CODENAME" == "xenial" ]; then # TODO(ARROW-4761): Install libzstd-dev once we support zstd<1 diff --git a/ci/travis_install_minio.sh b/ci/travis_install_minio.sh new file mode 100755 index 000000000000..5459bb8bfc4a --- /dev/null +++ b/ci/travis_install_minio.sh @@ -0,0 +1,35 @@ +#!/usr/bin/env bash + +# 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. + +set -e +set -x + +if [ "$ARROW_TRAVIS_S3" == "1" ]; then + # Download the Minio S3 server into PATH + if [ $TRAVIS_OS_NAME = "osx" ]; then + MINIO_URL=https://dl.min.io/server/minio/release/darwin-amd64/minio + else + MINIO_URL=https://dl.min.io/server/minio/release/linux-amd64/minio + fi + + S3FS_DIR=~/.local/bin/ + mkdir -p $S3FS_DIR + wget --directory-prefix $S3FS_DIR $MINIO_URL + chmod +x $S3FS_DIR/minio +fi \ No newline at end of file diff --git a/ci/travis_install_osx.sh b/ci/travis_install_osx.sh index 38e971710dea..2d79eb017ed5 100755 --- a/ci/travis_install_osx.sh +++ b/ci/travis_install_osx.sh @@ -40,4 +40,4 @@ if [ "$ARROW_CI_RUBY_AFFECTED" = "1" ]; then run_brew bundle --file=$TRAVIS_BUILD_DIR/cpp/Brewfile --verbose run_brew bundle --file=$TRAVIS_BUILD_DIR/c_glib/Brewfile --verbose rm ${brew_log_path} -fi +fi \ No newline at end of file diff --git a/ci/travis_script_python.sh b/ci/travis_script_python.sh index 202c24f0a58c..89fa4addae52 100755 --- a/ci/travis_script_python.sh +++ b/ci/travis_script_python.sh @@ -100,6 +100,10 @@ CMAKE_COMMON_FLAGS="-DARROW_EXTRA_ERROR_CONTEXT=ON" PYTHON_CPP_BUILD_TARGETS="arrow_python-all plasma parquet" +if [ "$ARROW_TRAVIS_S3" == "1" ]; then + CMAKE_COMMON_FLAGS="$CMAKE_COMMON_FLAGS -DARROW_S3=ON" +fi + if [ "$ARROW_TRAVIS_FLIGHT" == "1" ]; then CMAKE_COMMON_FLAGS="$CMAKE_COMMON_FLAGS -DARROW_FLIGHT=ON" fi @@ -164,6 +168,9 @@ export PYARROW_BUILD_TYPE=$ARROW_BUILD_TYPE export PYARROW_WITH_PARQUET=1 export PYARROW_WITH_PLASMA=1 export PYARROW_WITH_ORC=1 +if [ "$ARROW_TRAVIS_S3" == "1" ]; then + export PYARROW_WITH_S3=1 +fi if [ "$ARROW_TRAVIS_FLIGHT" == "1" ]; then export PYARROW_WITH_FLIGHT=1 fi @@ -177,6 +184,7 @@ python setup.py develop python -c "import pyarrow.parquet" python -c "import pyarrow.plasma" python -c "import pyarrow.orc" +python -c "import pyarrow.fs" # Ensure we do eagerly import pandas (or other expensive imports) python < scripts/test_imports.py diff --git a/cpp/cmake_modules/ThirdpartyToolchain.cmake b/cpp/cmake_modules/ThirdpartyToolchain.cmake index e8e4fa6f7bc7..49fb104502fd 100644 --- a/cpp/cmake_modules/ThirdpartyToolchain.cmake +++ b/cpp/cmake_modules/ThirdpartyToolchain.cmake @@ -2535,6 +2535,16 @@ 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) + # CoreFoundation's path is hardcoded in the CMake files provided by + # aws-sdk-cpp to use the MacOSX SDK provided by XCode which makes + # XCode a hard dependency. Command Line Tools is often used instead + # of the full XCode suite, so let the linker to find it. + set_target_properties(AWS::aws-c-common + PROPERTIES INTERFACE_LINK_LIBRARIES + "-pthread;pthread;-framework CoreFoundation") + endif() endif() # Write out the package configurations. diff --git a/cpp/src/arrow/filesystem/s3fs.cc b/cpp/src/arrow/filesystem/s3fs.cc index 136d12e20302..0e3c75ea7907 100644 --- a/cpp/src/arrow/filesystem/s3fs.cc +++ b/cpp/src/arrow/filesystem/s3fs.cc @@ -340,6 +340,12 @@ class ObjectInputFile : public io::RandomAccessFile { RETURN_NOT_OK(CheckClosed()); RETURN_NOT_OK(CheckPosition(position, "read")); + nbytes = std::min(nbytes, content_length_ - position); + if (nbytes == 0) { + *bytes_read = 0; + return Status::OK(); + } + // Read the desired range of bytes S3Model::GetObjectResult result; RETURN_NOT_OK(GetObjectRange(client_, path_, position, nbytes, &result)); diff --git a/cpp/src/arrow/filesystem/s3fs.h b/cpp/src/arrow/filesystem/s3fs.h index c02c4c154028..25720784622e 100644 --- a/cpp/src/arrow/filesystem/s3fs.h +++ b/cpp/src/arrow/filesystem/s3fs.h @@ -129,7 +129,7 @@ class ARROW_EXPORT S3FileSystem : public FileSystem { std::unique_ptr impl_; }; -enum class S3LogLevel { Off, Fatal, Error, Warn, Info, Debug, Trace }; +enum class S3LogLevel : int8_t { Off, Fatal, Error, Warn, Info, Debug, Trace }; struct ARROW_EXPORT S3GlobalOptions { S3LogLevel log_level; diff --git a/cpp/src/arrow/filesystem/s3fs_test.cc b/cpp/src/arrow/filesystem/s3fs_test.cc index b19b18e01638..304ae5a41b93 100644 --- a/cpp/src/arrow/filesystem/s3fs_test.cc +++ b/cpp/src/arrow/filesystem/s3fs_test.cc @@ -666,7 +666,7 @@ TEST_F(TestS3FS, OpenInputStream) { TEST_F(TestS3FS, OpenInputFile) { std::shared_ptr file; std::shared_ptr buf; - int64_t nbytes = -1, pos = -1; + int64_t nbytes = -1, pos = -1, bytes_read = 0; // Non-existent ASSERT_RAISES(IOError, fs_->OpenInputFile("non-existent-bucket/somefile", &file)); @@ -691,6 +691,15 @@ TEST_F(TestS3FS, OpenInputFile) { AssertBufferEqual(*buf, "data"); ASSERT_OK(file->ReadAt(9, 20, &buf)); AssertBufferEqual(*buf, ""); + + char result[10]; + ASSERT_OK(file->ReadAt(2, 5, &bytes_read, &result)); + ASSERT_EQ(bytes_read, 5); + ASSERT_OK(file->ReadAt(5, 20, &bytes_read, &result)); + ASSERT_EQ(bytes_read, 4); + ASSERT_OK(file->ReadAt(9, 0, &bytes_read, &result)); + ASSERT_EQ(bytes_read, 0); + // Reading past end of file ASSERT_RAISES(IOError, file->ReadAt(10, 20, &buf)); diff --git a/python/CMakeLists.txt b/python/CMakeLists.txt index b2282a6b69bc..6925efd2d17e 100644 --- a/python/CMakeLists.txt +++ b/python/CMakeLists.txt @@ -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} _s3fs) +endif() + if(PYARROW_BUILD_CUDA) # Arrow CUDA find_package(ArrowCuda) diff --git a/python/pyarrow/_csv.pyx b/python/pyarrow/_csv.pyx index 25ff47d39b54..5dccad760f42 100644 --- a/python/pyarrow/_csv.pyx +++ b/python/pyarrow/_csv.pyx @@ -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 diff --git a/python/pyarrow/_cuda.pyx b/python/pyarrow/_cuda.pyx index a9f51b0a654f..e0cad68b2bfe 100644 --- a/python/pyarrow/_cuda.pyx +++ b/python/pyarrow/_cuda.pyx @@ -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): diff --git a/python/pyarrow/_fs.pxd b/python/pyarrow/_fs.pxd new file mode 100644 index 000000000000..11b5769f854e --- /dev/null +++ b/python/pyarrow/_fs.pxd @@ -0,0 +1,68 @@ +# 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 * + + +cpdef enum FileType: + NonExistent = CFileType_NonExistent + Unknown = CFileType_Unknown + File = CFileType_File + Directory = CFileType_Directory + + +cdef class FileStats: + cdef: + CFileStats stats + + @staticmethod + cdef FileStats wrap(CFileStats stats) + + +cdef class Selector: + cdef: + CSelector selector + + +cdef class FileSystem: + cdef: + shared_ptr[CFileSystem] wrapped + CFileSystem* fs + + cdef init(self, const shared_ptr[CFileSystem]& wrapped) + + +cdef class LocalFileSystem(FileSystem): + cdef: + CLocalFileSystem* localfs + + cdef init(self, const shared_ptr[CFileSystem]& wrapped) + + +cdef class SubTreeFileSystem(FileSystem): + cdef: + CSubTreeFileSystem* subtreefs + + cdef init(self, const shared_ptr[CFileSystem]& wrapped) diff --git a/python/pyarrow/_fs.pyx b/python/pyarrow/_fs.pyx index 769ca8d7391f..39079aee1051 100644 --- a/python/pyarrow/_fs.pyx +++ b/python/pyarrow/_fs.pyx @@ -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 *: @@ -46,20 +37,12 @@ cdef inline c_string _path_as_bytes(path) except *: return tobytes(path) -cpdef enum FileType: - NonExistent = CFileType_NonExistent - Unknown = CFileType_Unknown - File = CFileType_File - Directory = CFileType_Directory - - cdef class FileStats: """FileSystem entry stats""" - cdef CFileStats stats - def __init__(self): - raise TypeError('dont initialize me') + raise TypeError("FileStats cannot be instantiated directly, use " + "FileSystem.get_target_stats method instead.") @staticmethod cdef FileStats wrap(CFileStats stats): @@ -115,9 +98,7 @@ cdef class FileStats: Only regular files are guaranteed to have a size. """ if self.stats.type() != CFileType_File: - raise ValueError( - 'Only regular files are guaranteed to have a size' - ) + return None return self.stats.size() @property @@ -146,7 +127,7 @@ cdef class Selector: Parameters ---------- - base_dir : str or pathlib.Path + base_dir : str The directory in which to select files. Relative paths also work, use '.' for the current directory and '..' for the parent. allow_non_existent : bool, default False @@ -156,7 +137,6 @@ cdef class Selector: recursive : bool, default False Whether to recurse into subdirectories. """ - cdef CSelector selector def __init__(self, base_dir, bint allow_non_existent=False, bint recursive=False): @@ -192,10 +172,6 @@ cdef class Selector: cdef class FileSystem: """Abstract file system API""" - cdef: - shared_ptr[CFileSystem] wrapped - CFileSystem* fs - def __init__(self): raise TypeError("FileSystem is an abstract class, instantiate one of " "the subclasses instead: LocalFileSystem or " @@ -249,7 +225,7 @@ cdef class FileSystem: Parameters ---------- - path : str or pathlib.Path + path : str The path of the new directory. recursive: bool, default True Create nested directories as well. @@ -263,7 +239,7 @@ cdef class FileSystem: Parameters ---------- - path : str or pathlib.Path + path : str The path of the directory to be deleted. """ cdef c_string directory = _path_as_bytes(path) @@ -280,9 +256,9 @@ cdef class FileSystem: Parameters ---------- - src : str or pathlib.Path + src : str The path of the file or the directory to be moved. - dest : str or pathlib.Path + dest : str The destination path where the file or directory is moved to. """ cdef: @@ -299,9 +275,9 @@ cdef class FileSystem: Parameters ---------- - src : str or pathlib.Path + src : str The path of the file to be copied from. - dest : str or pathlib.Path + dest : str The destination path where the file is copied to. """ cdef: @@ -315,7 +291,7 @@ cdef class FileSystem: Parameters ---------- - path : str or pathlib.Path + path : str The path of the file to be deleted. """ cdef c_string file = _path_as_bytes(path) @@ -345,7 +321,7 @@ cdef class FileSystem: Parameters ---------- - path : Union[str, pathlib.Path] + path : str The source to open for reading. Returns @@ -369,7 +345,7 @@ cdef class FileSystem: Parameters ---------- - source: str or pathlib.Path + source: str The source to open for reading. compression: str optional, default 'detect' The compression algorithm to use for on-the-fly decompression. @@ -407,7 +383,7 @@ cdef class FileSystem: Parameters ---------- - path : str or pathlib.Path + path : str The source to open for writing. compression: str optional, default 'detect' The compression algorithm to use for on-the-fly compression. @@ -445,7 +421,7 @@ cdef class FileSystem: Parameters ---------- - path : str or pathlib.Path + path : str The source to open for writing. compression: str optional, default 'detect' The compression algorithm to use for on-the-fly compression. @@ -484,9 +460,6 @@ cdef class LocalFileSystem(FileSystem): except when deleting an entry). """ - cdef: - CLocalFileSystem* localfs - def __init__(self): cdef shared_ptr[CLocalFileSystem] wrapped wrapped = make_shared[CLocalFileSystem]() @@ -506,10 +479,14 @@ cdef class SubTreeFileSystem(FileSystem): Note, that this makes no security guarantee. For example, symlinks may allow to "escape" the subtree and access other parts of the underlying filesystem. - """ - cdef: - CSubTreeFileSystem* subtreefs + Parameters + ---------- + base_path: str + The root of the subtree. + base_fs: FileSystem + FileSystem object the operations delegated to. + """ def __init__(self, base_path, FileSystem base_fs): cdef: diff --git a/python/pyarrow/_json.pyx b/python/pyarrow/_json.pyx index ffbf01c09e7c..da3588a5e625 100644 --- a/python/pyarrow/_json.pyx +++ b/python/pyarrow/_json.pyx @@ -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 diff --git a/python/pyarrow/_orc.pxd b/python/pyarrow/_orc.pxd index ebbf8beda828..649fe8248f41 100644 --- a/python/pyarrow/_orc.pxd +++ b/python/pyarrow/_orc.pxd @@ -28,7 +28,7 @@ from pyarrow.includes.libarrow cimport (CArray, CSchema, CStatus, CKeyValueMetadata, CRecordBatch, CTable, - RandomAccessFile, OutputStream, + CRandomAccessFile, 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) diff --git a/python/pyarrow/_orc.pyx b/python/pyarrow/_orc.pyx index c9f5b2e158d6..0ee3ca632e46 100644 --- a/python/pyarrow/_orc.pyx +++ b/python/pyarrow/_orc.pyx @@ -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 diff --git a/python/pyarrow/_parquet.pxd b/python/pyarrow/_parquet.pxd index 19fb214c8f04..5cf3ff5931b4 100644 --- a/python/pyarrow/_parquet.pxd +++ b/python/pyarrow/_parquet.pxd @@ -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) diff --git a/python/pyarrow/_parquet.pyx b/python/pyarrow/_parquet.pyx index 3d62c0eb13e6..f93def8a67ee 100644 --- a/python/pyarrow/_parquet.pyx +++ b/python/pyarrow/_parquet.pyx @@ -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: diff --git a/python/pyarrow/_s3fs.pyx b/python/pyarrow/_s3fs.pyx new file mode 100644 index 000000000000..d1f820e51958 --- /dev/null +++ b/python/pyarrow/_s3fs.pyx @@ -0,0 +1,173 @@ +# 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.lib cimport check_status +from pyarrow.compat import frombytes, tobytes +from pyarrow.includes.common cimport * +from pyarrow.includes.libarrow cimport * +from pyarrow.includes.libarrow_s3fs cimport * +from pyarrow._fs cimport FileSystem + + +cpdef enum S3LogLevel: + Off = CS3LogLevel_Off + Fatal = CS3LogLevel_Fatal + Error = CS3LogLevel_Error + Warn = CS3LogLevel_Warn + Info = CS3LogLevel_Info + Debug = CS3LogLevel_Debug + Trace = CS3LogLevel_Trace + + +def initialize_s3(S3LogLevel log_level=S3LogLevel.Error): + cdef CS3GlobalOptions options + options.log_level = log_level + check_status(CInitializeS3(options)) + + +def finalize_s3(): + check_status(CFinalizeS3()) + + +cdef class S3Options: + """Options for S3FileSystem. + + If neither access_key nor secret_key are provided then attempts to + initialize from AWS environment variables, otherwise both access_key and + secret_key must be provided. + + Parameters + ---------- + access_key: str, default None + AWS Access Key ID. Pass None to use the standard AWS environment + variables and/or configuration file. + secret_key: str, default None + AWS Secret Access key. Pass None to use the standard AWS environment + variables and/or configuration file. + region: str, default 'us-east-1' + AWS region to connect to. + scheme: str, default 'https' + S3 connection transport scheme. + endpoint_override: str, default None + Override region with a connect string such as "localhost:9000" + background_writes: boolean, default True + Whether OutputStream writes will be issued in the background, without + blocking. + """ + cdef: + CS3Options options + + # Avoid mistakingly creating attributes + __slots__ = () + + def __init__(self, access_key=None, secret_key=None, region=None, + scheme=None, endpoint_override=None, background_writes=None): + if access_key is not None and secret_key is None: + raise ValueError( + 'In order to initialize with explicit credentials both ' + 'access_key and secret_key must be provided, ' + '`secret_key` is not set.' + ) + elif access_key is None and secret_key is not None: + raise ValueError( + 'In order to initialize with explicit credentials both ' + 'access_key and secret_key must be provided, ' + '`access_key` is not set.' + ) + elif access_key is not None or secret_key is not None: + self.options = CS3Options.FromAccessKey( + tobytes(access_key), + tobytes(secret_key) + ) + else: + self.options = CS3Options.Defaults() + + if region is not None: + self.region = region + if scheme is not None: + self.scheme = scheme + if endpoint_override is not None: + self.endpoint_override = endpoint_override + if background_writes is not None: + self.background_writes = background_writes + + @property + def region(self): + """AWS region to connect to.""" + return frombytes(self.options.region) + + @region.setter + def region(self, value): + self.options.region = tobytes(value) + + @property + def scheme(self): + """S3 connection transport scheme.""" + return frombytes(self.options.scheme) + + @scheme.setter + def scheme(self, value): + self.options.scheme = tobytes(value) + + @property + def endpoint_override(self): + """Override region with a connect string such as localhost:9000""" + return frombytes(self.options.endpoint_override) + + @endpoint_override.setter + def endpoint_override(self, value): + self.options.endpoint_override = tobytes(value) + + @property + def background_writes(self): + """OutputStream writes will be issued in the background""" + return self.options.background_writes + + @background_writes.setter + def background_writes(self, bint value): + self.options.background_writes = value + + +cdef class S3FileSystem(FileSystem): + """S3-backed FileSystem implementation + + Note: S3 buckets are special and the operations available on them may be + limited or more expensive than desired. + + Parameters + ---------- + options: S3Options, default None + Options for connecting to S3. If None is passed then attempts to + initialize the connection from AWS environment variables. + """ + + cdef: + CS3FileSystem* s3fs + + def __init__(self, S3Options options=None): + cdef shared_ptr[CS3FileSystem] wrapped + options = options or S3Options() + check_status(CS3FileSystem.Make(options.options, &wrapped)) + self.init( wrapped) + + cdef init(self, const shared_ptr[CFileSystem]& wrapped): + FileSystem.init(self, wrapped) + self.s3fs = wrapped.get() diff --git a/python/pyarrow/feather.pxi b/python/pyarrow/feather.pxi index 6fd13bc04b46..8700f67ae621 100644 --- a/python/pyarrow/feather.pxi +++ b/python/pyarrow/feather.pxi @@ -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: diff --git a/python/pyarrow/fs.py b/python/pyarrow/fs.py index cd5263acbcad..5f257d07f300 100644 --- a/python/pyarrow/fs.py +++ b/python/pyarrow/fs.py @@ -17,4 +17,11 @@ from __future__ import absolute_import -from pyarrow._fs import * # noqa +from pyarrow._fs import ( # noqa + Selector, + FileType, + FileStats, + FileSystem, + LocalFileSystem, + SubTreeFileSystem +) diff --git a/python/pyarrow/includes/libarrow.pxd b/python/pyarrow/includes/libarrow.pxd index 69dafa4e46f2..82085487eda3 100644 --- a/python/pyarrow/includes/libarrow.pxd +++ b/python/pyarrow/includes/libarrow.pxd @@ -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,14 @@ 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 +940,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 +962,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: + + ctypedef 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 +1070,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 +1089,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 +1100,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 +1127,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 +1151,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 +1171,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 +1182,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 +1241,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 +1269,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 +1448,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 +1467,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 +1477,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, diff --git a/python/pyarrow/includes/libarrow_fs.pxd b/python/pyarrow/includes/libarrow_fs.pxd deleted file mode 100644 index f54a2e50357e..000000000000 --- a/python/pyarrow/includes/libarrow_fs.pxd +++ /dev/null @@ -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) diff --git a/python/pyarrow/includes/libarrow_s3fs.pxd b/python/pyarrow/includes/libarrow_s3fs.pxd new file mode 100644 index 000000000000..8dc109c5e6e3 --- /dev/null +++ b/python/pyarrow/includes/libarrow_s3fs.pxd @@ -0,0 +1,60 @@ +# 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: + + ctypedef enum CS3LogLevel "arrow::fs::S3LogLevel": + CS3LogLevel_Off "arrow::fs::S3LogLevel::Off" + CS3LogLevel_Fatal "arrow::fs::S3LogLevel::Fatal" + CS3LogLevel_Error "arrow::fs::S3LogLevel::Error" + CS3LogLevel_Warn "arrow::fs::S3LogLevel::Warn" + CS3LogLevel_Info "arrow::fs::S3LogLevel::Info" + CS3LogLevel_Debug "arrow::fs::S3LogLevel::Debug" + CS3LogLevel_Trace "arrow::fs::S3LogLevel::Trace" + + cdef struct CS3GlobalOptions "arrow::fs::S3GlobalOptions": + CS3LogLevel log_level + + cdef cppclass CS3Options "arrow::fs::S3Options": + c_string region + c_string endpoint_override + c_string scheme + c_bool background_writes + void ConfigureDefaultCredentials() + void ConfigureAccessKey(const c_string& access_key, + const c_string& secret_key) + + @staticmethod + CS3Options Defaults() + @staticmethod + CS3Options FromAccessKey(const c_string& access_key, + const c_string& secret_key) + + cdef cppclass CS3FileSystem "arrow::fs::S3FileSystem"(CFileSystem): + @staticmethod + CStatus Make(const CS3Options& options, shared_ptr[CS3FileSystem]* out) + + cdef CStatus CInitializeS3 "arrow::fs::InitializeS3"( + const CS3GlobalOptions& options) + cdef CStatus CFinalizeS3 "arrow::fs::FinalizeS3"() diff --git a/python/pyarrow/io-hdfs.pxi b/python/pyarrow/io-hdfs.pxi index e9655e4a7f74..b224abbead82 100644 --- a/python/pyarrow/io-hdfs.pxi +++ b/python/pyarrow/io-hdfs.pxi @@ -424,7 +424,7 @@ cdef class HadoopFileSystem: c_replication, c_default_block_size, &wr_handle)) - out.set_output_stream( wr_handle) + out.set_output_stream( 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( - rd_handle) + rd_handle) out.is_readable = True assert not out.closed diff --git a/python/pyarrow/io.pxi b/python/pyarrow/io.pxi index 6ff1403006ef..0cdf21efc010 100644 --- a/python/pyarrow/io.pxi +++ b/python/pyarrow/io.pxi @@ -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 = handle + cdef set_random_access_file(self, shared_ptr[CRandomAccessFile] handle): + self.input_stream = 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( handle) - result.set_random_access_file( handle) + result.set_output_stream( handle) + result.set_random_access_file( 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( handle) - self.set_random_access_file( handle) + self.set_output_stream( handle) + self.set_random_access_file( 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( handle) + self.set_random_access_file( 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( compressed_stream) + self.set_input_stream( 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( compressed_stream) + self.set_output_stream( 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( buffered_stream) + self.set_input_stream( 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( buffered_stream) + self.set_output_stream( 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 = compressed_stream + input_stream = 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: diff --git a/python/pyarrow/ipc.pxi b/python/pyarrow/ipc.pxi index 3d8b3f4af4db..c9684f13e5bb 100644 --- a/python/pyarrow/ipc.pxi +++ b/python/pyarrow/ipc.pxi @@ -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 diff --git a/python/pyarrow/lib.pxd b/python/pyarrow/lib.pxd index 553227a4e559..571dfaa9449d 100644 --- a/python/pyarrow/lib.pxd +++ b/python/pyarrow/lib.pxd @@ -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=*) diff --git a/python/pyarrow/s3fs.py b/python/pyarrow/s3fs.py new file mode 100644 index 000000000000..5619e186f9ea --- /dev/null +++ b/python/pyarrow/s3fs.py @@ -0,0 +1,27 @@ +# 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. + +from __future__ import absolute_import + +from pyarrow._s3fs import ( # noqa + initialize_s3, + finalize_s3, + S3Options, + S3FileSystem +) + +initialize_s3() diff --git a/python/pyarrow/serialization.pxi b/python/pyarrow/serialization.pxi index fba834434d02..cf343b0ba320 100644 --- a/python/pyarrow/serialization.pxi +++ b/python/pyarrow/serialization.pxi @@ -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() diff --git a/python/pyarrow/tests/conftest.py b/python/pyarrow/tests/conftest.py index a095d81bcf70..68e215b3f422 100644 --- a/python/pyarrow/tests/conftest.py +++ b/python/pyarrow/tests/conftest.py @@ -16,6 +16,9 @@ # under the License. import os +import subprocess +import tempfile + import pytest import hypothesis as h @@ -24,6 +27,8 @@ except ImportError: import pathlib2 as pathlib # py2 compat +from pyarrow.util import find_free_port + # setup hypothesis profiles h.settings.register_profile('ci', max_examples=1000) @@ -126,6 +131,12 @@ except ImportError: pass +try: + import pyarrow.s3fs # noqa + defaults['s3'] = True +except ImportError: + pass + def pytest_configure(config): for mark in groups: @@ -219,3 +230,47 @@ def tempdir(tmpdir): @pytest.fixture(scope='session') def datadir(): return pathlib.Path(__file__).parent / 'data' + + +try: + from tempfile import TemporaryDirectory +except ImportError: + import shutil + + class TemporaryDirectory(object): + """Temporary directory implementation for python 2""" + + def __enter__(self): + self.tmp = tempfile.mkdtemp() + return self.tmp + + def __exit__(self, exc_type, exc_value, traceback): + shutil.rmtree(self.tmp) + + +@pytest.mark.s3 +@pytest.fixture(scope='session') +def minio_server(): + host, port = 'localhost', find_free_port() + access_key, secret_key = 'arrow', 'apachearrow' + + address = '{}:{}'.format(host, port) + env = os.environ.copy() + env.update({ + 'MINIO_ACCESS_KEY': access_key, + 'MINIO_SECRET_KEY': secret_key + }) + + with TemporaryDirectory() as tempdir: + args = ['minio', '--compat', 'server', '--quiet', '--address', + address, tempdir] + proc = None + try: + proc = subprocess.Popen(args, env=env) + except IOError: + pytest.skip('`minio` command cannot be located') + else: + yield address, access_key, secret_key + finally: + if proc is not None: + proc.kill() diff --git a/python/pyarrow/tests/test_flight.py b/python/pyarrow/tests/test_flight.py index 83afba6de65e..40099b06aec6 100644 --- a/python/pyarrow/tests/test_flight.py +++ b/python/pyarrow/tests/test_flight.py @@ -17,9 +17,7 @@ # under the License. import base64 -import contextlib import os -import socket import struct import tempfile import threading @@ -30,7 +28,7 @@ import pyarrow as pa from pyarrow.compat import tobytes -from pyarrow.util import pathlib +from pyarrow.util import pathlib, find_free_port try: from pyarrow import flight @@ -48,14 +46,6 @@ pytestmark = pytest.mark.flight -def find_free_port(): - sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) - with contextlib.closing(sock) as sock: - sock.bind(('', 0)) - sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) - return sock.getsockname()[1] - - def test_import(): # So we see the ImportError somewhere import pyarrow.flight # noqa diff --git a/python/pyarrow/tests/test_fs.py b/python/pyarrow/tests/test_fs.py index f897e0d36f97..f6b6bf1d18cd 100644 --- a/python/pyarrow/tests/test_fs.py +++ b/python/pyarrow/tests/test_fs.py @@ -23,39 +23,109 @@ import pytest -from pyarrow import ArrowIOError +import pyarrow as pa +from pyarrow.tests.test_io import gzip_compress, gzip_decompress from pyarrow.fs import (FileType, Selector, FileSystem, LocalFileSystem, SubTreeFileSystem) -from pyarrow.tests.test_io import gzip_compress, gzip_decompress + + +@pytest.fixture +def localfs(request, tempdir): + return dict( + fs=LocalFileSystem(), + pathfn=lambda p: (tempdir / p).as_posix(), + allow_move_dir=True, + allow_append_to_file=True, + ) + + +@pytest.fixture +def subtree_localfs(request, tempdir, localfs): + prefix = 'subtree/prefix/' + (tempdir / prefix).mkdir(parents=True) + return dict( + fs=SubTreeFileSystem(prefix, localfs['fs']), + pathfn=prefix.__add__, + allow_move_dir=True, + allow_append_to_file=True, + ) + + +@pytest.mark.s3 +@pytest.fixture +def s3fs(request, minio_server): + from pyarrow.s3fs import S3Options, S3FileSystem + + address, access_key, secret_key = minio_server + bucket = 'pyarrow-filesystem/' + options = S3Options( + endpoint_override=address, + access_key=access_key, + secret_key=secret_key, + scheme='http' + ) + fs = S3FileSystem(options) + fs.create_dir(bucket) + + return dict( + fs=fs, + pathfn=bucket.__add__, + allow_move_dir=False, + allow_append_to_file=False, + ) + + +@pytest.fixture +def subtree_s3fs(request, s3fs): + prefix = 'pyarrow-filesystem/prefix/' + return dict( + fs=SubTreeFileSystem(prefix, s3fs['fs']), + pathfn=prefix.__add__, + allow_move_dir=False, + allow_append_to_file=False, + ) @pytest.fixture(params=[ pytest.param( - lambda tmp: LocalFileSystem(), - id='LocalFileSystem' + pytest.lazy_fixture('localfs'), + id='LocalFileSystem()' ), pytest.param( - lambda tmp: SubTreeFileSystem(tmp, LocalFileSystem()), - id='SubTreeFileSystem(LocalFileSystem)' + pytest.lazy_fixture('subtree_localfs'), + id='SubTreeFileSystem(LocalFileSystem())' + ), + pytest.param( + pytest.lazy_fixture('s3fs'), + id='S3FileSystem' + ), + pytest.param( + pytest.lazy_fixture('subtree_s3fs'), + id='SubTreeFileSystem(S3FileSystem())' ) ]) -def fs(request, tempdir): - return request.param(tempdir.as_posix()) +def filesystem_config(request): + return request.param + + +@pytest.fixture +def fs(request, filesystem_config): + return filesystem_config['fs'] @pytest.fixture -def testpath(request, fs, tempdir): - # we always use the tempdir for reading and writing test artifacts, but - # if the filesystem is wrapped in a SubTreeFileSystem then we don't need - # to prepend the path with the tempdir, we also test the API with both - # pathlib.Path objects and plain python strings - def convert(path): - if isinstance(fs, SubTreeFileSystem): - path = pathlib.Path(path) - else: - path = tempdir / path - return path.as_posix() - return convert +def pathfn(request, filesystem_config): + return filesystem_config['pathfn'] + + +@pytest.fixture +def allow_move_dir(request, filesystem_config): + return filesystem_config['allow_move_dir'] + + +@pytest.fixture +def allow_append_to_file(request, filesystem_config): + return filesystem_config['allow_append_to_file'] def test_cannot_instantiate_base_filesystem(): @@ -74,157 +144,158 @@ class Path: fs.create_dir(path) -def test_get_target_stats(fs, tempdir, testpath): - aaa, aaa_ = testpath('a/aa/aaa'), tempdir / 'a' / 'aa' / 'aaa' - bb, bb_ = testpath('a/bb'), tempdir / 'a' / 'bb' - c, c_ = testpath('c.txt'), tempdir / 'c.txt' - - aaa_.mkdir(parents=True) - bb_.touch() - c_.write_bytes(b'test') +def test_get_target_stats(fs, pathfn): + aaa = pathfn('a/aa/aaa/') + bb = pathfn('a/bb') + c = pathfn('c.txt') - def mtime_almost_equal(fs_dt, pathlib_ts): - # arrow's filesystem implementation truncates mtime to microsends - # resolution whereas pathlib rounds - pathlib_dt = datetime.utcfromtimestamp(pathlib_ts) - difference = (fs_dt - pathlib_dt).total_seconds() - return abs(difference) <= 10**-6 + fs.create_dir(aaa) + with fs.open_output_stream(bb): + pass # touch + with fs.open_output_stream(c) as fp: + fp.write(b'test') aaa_stat, bb_stat, c_stat = fs.get_target_stats([aaa, bb, c]) assert aaa_stat.path == aaa assert 'aaa' in repr(aaa_stat) - assert aaa_stat.base_name == 'aaa' assert aaa_stat.extension == '' - assert aaa_stat.type == FileType.Directory - assert mtime_almost_equal(aaa_stat.mtime, aaa_.stat().st_mtime) - with pytest.raises(ValueError): - aaa_stat.size + assert isinstance(aaa_stat.mtime, datetime) assert bb_stat.path == str(bb) assert bb_stat.base_name == 'bb' assert bb_stat.extension == '' assert bb_stat.type == FileType.File assert bb_stat.size == 0 - assert mtime_almost_equal(bb_stat.mtime, bb_.stat().st_mtime) + assert isinstance(bb_stat.mtime, datetime) assert c_stat.path == str(c) assert c_stat.base_name == 'c.txt' assert c_stat.extension == 'txt' assert c_stat.type == FileType.File assert c_stat.size == 4 - assert mtime_almost_equal(c_stat.mtime, c_.stat().st_mtime) - - -def test_get_target_stats_with_selector(fs, tempdir, testpath): - base_dir = testpath('.') - base_dir_ = tempdir - - selector = Selector(base_dir, allow_non_existent=False, recursive=True) - assert selector.base_dir == str(base_dir) - - (tempdir / 'test_file').touch() - (tempdir / 'test_directory').mkdir() - - stats = fs.get_target_stats(selector) - expected = list(base_dir_.iterdir()) - assert len(stats) == len(expected) - - for st in stats: - p = base_dir_ / st.path - if p.is_dir(): - assert st.type == FileType.Directory - if p.is_file(): - assert st.type == FileType.File - - -def test_create_dir(fs, tempdir, testpath): - directory = testpath('directory') - directory_ = tempdir / 'directory' - assert not directory_.exists() - fs.create_dir(directory) - assert directory_.exists() - - # recursive - directory = testpath('deeply/nested/directory') - directory_ = tempdir / 'deeply' / 'nested' / 'directory' - assert not directory_.exists() - with pytest.raises(ArrowIOError): - fs.create_dir(directory, recursive=False) - fs.create_dir(directory) - assert directory_.exists() - - -def test_delete_dir(fs, tempdir, testpath): - folder = testpath('directory') - nested = testpath('nested/directory') - folder_ = tempdir / 'directory' - nested_ = tempdir / 'nested' / 'directory' - - folder_.mkdir() - nested_.mkdir(parents=True) - - assert folder_.exists() - fs.delete_dir(folder) - assert not folder_.exists() - - assert nested_.exists() - fs.delete_dir(nested) - assert not nested_.exists() - - -def test_copy_file(fs, tempdir, testpath): - # copy file - source = testpath('source-file') - source_ = tempdir / 'source-file' - source_.touch() - target = testpath('target-file') - target_ = tempdir / 'target-file' - assert not target_.exists() - fs.copy_file(source, target) - assert source_.exists() - assert target_.exists() - - -def test_move(fs, tempdir, testpath): - # move directory - source = testpath('source-dir') - source_ = tempdir / 'source-dir' - source_.mkdir() - target = testpath('target-dir') - target_ = tempdir / 'target-dir' - assert not target_.exists() - fs.move(source, target) - assert not source_.exists() - assert target_.exists() - - # move file - source = testpath('source-file') - source_ = tempdir / 'source-file' - source_.touch() - target = testpath('target-file') - target_ = tempdir / 'target-file' - assert not target_.exists() - fs.move(source, target) - assert not source_.exists() - assert target_.exists() - - -def test_delete_file(fs, tempdir, testpath): - target = testpath('target-file') - target_ = tempdir / 'target-file' - target_.touch() - assert target_.exists() - fs.delete_file(target) - assert not target_.exists() - - nested = testpath('nested/target-file') - nested_ = tempdir / 'nested/target-file' - nested_.parent.mkdir() - nested_.touch() - assert nested_.exists() - fs.delete_file(nested) - assert not nested_.exists() + assert isinstance(c_stat.mtime, datetime) + + +def test_get_target_stats_with_selector(fs, pathfn): + base_dir = pathfn('selector-dir/') + file_a = pathfn('selector-dir/test_file_a') + file_b = pathfn('selector-dir/test_file_b') + dir_a = pathfn('selector-dir/test_dir_a') + + try: + fs.create_dir(base_dir) + with fs.open_output_stream(file_a): + pass + with fs.open_output_stream(file_b): + pass + fs.create_dir(dir_a) + + selector = Selector(base_dir, allow_non_existent=False, recursive=True) + assert selector.base_dir == base_dir + + stats = fs.get_target_stats(selector) + assert len(stats) == 3 + + for st in stats: + if st.path.endswith(file_a): + assert st.type == FileType.File + elif st.path.endswith(file_b): + assert st.type == FileType.File + elif st.path.endswith(dir_a): + assert st.type == FileType.Directory + else: + raise ValueError('unexpected path {}'.format(st.path)) + finally: + fs.delete_file(file_a) + fs.delete_file(file_b) + fs.delete_dir(dir_a) + fs.delete_dir(base_dir) + + +def test_create_dir(fs, pathfn): + d = pathfn('test-directory/') + + with pytest.raises(pa.ArrowIOError): + fs.delete_dir(d) + + fs.create_dir(d) + fs.delete_dir(d) + + d = pathfn('deeply/nested/test-directory/') + fs.create_dir(d, recursive=True) + fs.delete_dir(d) + + +def test_delete_dir(fs, pathfn): + d = pathfn('directory/') + nd = pathfn('directory/nested/') + + fs.create_dir(nd) + fs.delete_dir(nd) + fs.delete_dir(d) + with pytest.raises(pa.ArrowIOError): + fs.delete_dir(d) + + +def test_copy_file(fs, pathfn): + s = pathfn('test-copy-source-file') + t = pathfn('test-copy-target-file') + + with fs.open_output_stream(s): + pass + + fs.copy_file(s, t) + fs.delete_file(s) + fs.delete_file(t) + + +def test_move_directory(fs, pathfn, allow_move_dir): + # move directory (doesn't work with S3) + s = pathfn('source-dir/') + t = pathfn('target-dir/') + + fs.create_dir(s) + + if allow_move_dir: + fs.move(s, t) + with pytest.raises(pa.ArrowIOError): + fs.delete_dir(s) + fs.delete_dir(t) + else: + with pytest.raises(pa.ArrowIOError): + fs.move(s, t) + + +def test_move_file(fs, pathfn): + s = pathfn('test-move-source-file') + t = pathfn('test-move-target-file') + + with fs.open_output_stream(s): + pass + + fs.move(s, t) + with pytest.raises(pa.ArrowIOError): + fs.delete_file(s) + fs.delete_file(t) + + +def test_delete_file(fs, pathfn): + p = pathfn('test-delete-target-file') + with fs.open_output_stream(p): + pass + + fs.delete_file(p) + with pytest.raises(pa.ArrowIOError): + fs.delete_file(p) + + d = pathfn('test-delete-nested') + fs.create_dir(d) + f = pathfn('test-delete-nested/target-file') + with fs.open_output_stream(f) as s: + s.write(b'data') + + fs.delete_dir(d) def identity(v): @@ -240,27 +311,28 @@ def identity(v): ('gzip', 256, gzip_compress), ] ) -def test_open_input_stream(fs, tempdir, testpath, compression, buffer_size, - compressor): - file = testpath('abc') - file_ = tempdir / 'abc' - data = b'some data' * 1024 - file_.write_bytes(compressor(data)) +def test_open_input_stream(fs, pathfn, compression, buffer_size, compressor): + p = pathfn('open-input-stream') - with fs.open_input_stream(file, compression, buffer_size) as f: - result = f.read() + data = b'some data for reading\n' * 512 + with fs.open_output_stream(p) as s: + s.write(compressor(data)) + + with fs.open_input_stream(p, compression, buffer_size) as s: + result = s.read() assert result == data -def test_open_input_file(fs, tempdir, testpath): - file = testpath('abc') - file_ = tempdir / 'abc' +def test_open_input_file(fs, pathfn): + p = pathfn('open-input-file') + data = b'some data' * 1024 - file_.write_bytes(data) + with fs.open_output_stream(p) as s: + s.write(data) read_from = len(b'some data') * 512 - with fs.open_input_file(file) as f: + with fs.open_input_file(p) as f: f.seek(read_from) result = f.read() @@ -276,16 +348,16 @@ def test_open_input_file(fs, tempdir, testpath): ('gzip', 256, gzip_decompress), ] ) -def test_open_output_stream(fs, tempdir, testpath, compression, buffer_size, +def test_open_output_stream(fs, pathfn, compression, buffer_size, decompressor): - file = testpath('abc') - file_ = tempdir / 'abc' + p = pathfn('open-output-stream') - data = b'some data' * 1024 - with fs.open_output_stream(file, compression, buffer_size) as f: + data = b'some data for writing' * 1024 + with fs.open_output_stream(p, compression, buffer_size) as f: f.write(data) - assert decompressor(file_.read_bytes()) == data + with fs.open_input_stream(p, compression, buffer_size) as f: + assert f.read(len(data)) == data @pytest.mark.parametrize( @@ -297,13 +369,57 @@ def test_open_output_stream(fs, tempdir, testpath, compression, buffer_size, ('gzip', 256, gzip_compress, gzip_decompress), ] ) -def test_open_append_stream(fs, tempdir, testpath, compression, buffer_size, - compressor, decompressor): - file = testpath('abc') - file_ = tempdir / 'abc' - file_.write_bytes(compressor(b'already existing')) +def test_open_append_stream(fs, pathfn, compression, buffer_size, compressor, + decompressor, allow_append_to_file): + p = pathfn('open-append-stream') + + initial = compressor(b'already existing') + with fs.open_output_stream(p) as s: + s.write(initial) + + if allow_append_to_file: + with fs.open_append_stream(p, compression, buffer_size) as f: + f.write(b'\nnewly added') - with fs.open_append_stream(file, compression, buffer_size) as f: - f.write(b'\nnewly added') + with fs.open_input_stream(p) as f: + result = f.read() - assert decompressor(file_.read_bytes()) == b'already existing\nnewly added' + result = decompressor(result) + assert result == b'already existing\nnewly added' + else: + with pytest.raises(pa.ArrowNotImplementedError): + fs.open_append_stream(p, compression, buffer_size) + + +@pytest.mark.s3 +def test_s3_options(minio_server): + from pyarrow.s3fs import S3Options + + options = S3Options() + + assert options.region == 'us-east-1' + options.region = 'us-west-1' + assert options.region == 'us-west-1' + + assert options.scheme == 'https' + options.scheme = 'http' + assert options.scheme == 'http' + + assert options.endpoint_override == '' + options.endpoint_override = 'localhost:8999' + assert options.endpoint_override == 'localhost:8999' + + with pytest.raises(ValueError): + S3Options(access_key='access') + with pytest.raises(ValueError): + S3Options(secret_key='secret') + + address, access_key, secret_key = minio_server + options = S3Options( + access_key=access_key, + secret_key=secret_key, + endpoint_override=address, + scheme='http' + ) + assert options.scheme == 'http' + assert options.endpoint_override == address diff --git a/python/pyarrow/tests/test_parquet.py b/python/pyarrow/tests/test_parquet.py index f8a3563e25f9..566f22f43110 100644 --- a/python/pyarrow/tests/test_parquet.py +++ b/python/pyarrow/tests/test_parquet.py @@ -1843,18 +1843,41 @@ def test_filters_read_table(tempdir): assert table.num_rows == 3 -@pytest.yield_fixture -def s3_example(): - access_key = os.environ['PYARROW_TEST_S3_ACCESS_KEY'] - secret_key = os.environ['PYARROW_TEST_S3_SECRET_KEY'] - bucket_name = os.environ['PYARROW_TEST_S3_BUCKET'] +@pytest.fixture +def s3_bucket(request, minio_server): + boto3 = pytest.importorskip('boto3') + botocore = pytest.importorskip('botocore') + + address, access_key, secret_key = minio_server + s3 = boto3.resource( + 's3', + endpoint_url='http://{}'.format(address), + aws_access_key_id=access_key, + aws_secret_access_key=secret_key, + config=botocore.client.Config(signature_version='s3v4'), + region_name='us-east-1' + ) + bucket = s3.Bucket('test-s3fs') + bucket.create() + return 'test-s3fs' + - import s3fs - fs = s3fs.S3FileSystem(key=access_key, secret=secret_key) +@pytest.fixture +def s3_example(minio_server, s3_bucket): + s3fs = pytest.importorskip('s3fs') + + address, access_key, secret_key = minio_server + fs = s3fs.S3FileSystem( + key=access_key, + secret=secret_key, + client_kwargs={ + 'endpoint_url': 'http://{}'.format(address) + } + ) test_dir = guid() + bucket_uri = 's3://{0}/{1}'.format(s3_bucket, test_dir) - bucket_uri = 's3://{0}/{1}'.format(bucket_name, test_dir) fs.mkdir(bucket_uri) yield fs, bucket_uri fs.rm(bucket_uri, recursive=True) @@ -1920,23 +1943,29 @@ def _visit_level(base_dir, level, part_keys): for value in values: this_part_keys = part_keys + [(name, value)] - level_dir = base_dir / '{0}={1}'.format(name, value) + level_dir = fs._path_join( + str(base_dir), + '{0}={1}'.format(name, value) + ) fs.mkdir(level_dir) if level == DEPTH - 1: # Generate example data - file_path = level_dir / guid() - + file_path = fs._path_join(level_dir, guid()) filtered_df = _filter_partition(df, this_part_keys) part_table = pa.Table.from_pandas(filtered_df) with fs.open(file_path, 'wb') as f: _write_table(part_table, f) assert fs.exists(file_path) - (level_dir / '_SUCCESS').touch() + file_success = fs._path_join(level_dir, '_SUCCESS') + with fs.open(file_success, 'wb') as f: + pass else: _visit_level(level_dir, level + 1, this_part_keys) - (level_dir / '_SUCCESS').touch() + file_success = fs._path_join(level_dir, '_SUCCESS') + with fs.open(file_success, 'wb') as f: + pass _visit_level(base_dir, 0, []) diff --git a/python/pyarrow/util.py b/python/pyarrow/util.py index 5e4fb3579372..7219a447f356 100644 --- a/python/pyarrow/util.py +++ b/python/pyarrow/util.py @@ -19,8 +19,10 @@ from __future__ import absolute_import +import contextlib import functools import six +import socket import warnings @@ -125,3 +127,11 @@ def get_contiguous_span(shape, strides, itemsize): if end - start != itemsize * product(shape): raise ValueError('array data is non-contiguous') return start, end + + +def find_free_port(): + sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + with contextlib.closing(sock) as sock: + sock.bind(('', 0)) + sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) + return sock.getsockname()[1] diff --git a/python/requirements-test.txt b/python/requirements-test.txt index 73eabfebd288..10d445cbc442 100644 --- a/python/requirements-test.txt +++ b/python/requirements-test.txt @@ -3,4 +3,5 @@ hypothesis pandas pathlib2; python_version < "3.4" pytest +pytest-lazy-fixture pytz diff --git a/python/setup.py b/python/setup.py index 5e88352a8f9b..d7207eedd8f1 100755 --- a/python/setup.py +++ b/python/setup.py @@ -141,6 +141,8 @@ def initialize_options(self): 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( @@ -176,6 +178,7 @@ def initialize_options(self): '_parquet', '_orc', '_plasma', + '_s3fs', 'gandiva'] def _run_cmake(self): @@ -215,6 +218,8 @@ def _run_cmake(self): 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 +419,8 @@ def _failure_permitted(self, name): return True if name == '_flight' and not self.with_flight: return True + if name == '_s3fs' 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: