ARROW-6655: [Python] Filesystem bindings for S3

.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

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

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

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

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

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

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

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.

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));

cpp/src/arrow/filesystem/s3fs.h

@@ -129,7 +129,7 @@ class ARROW_EXPORT S3FileSystem : public FileSystem {
   std::unique_ptr<Impl> 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;

cpp/src/arrow/filesystem/s3fs_test.cc

@@ -666,7 +666,7 @@ TEST_F(TestS3FS, OpenInputStream) {
 TEST_F(TestS3FS, OpenInputFile) {
   std::shared_ptr<io::RandomAccessFile> file;
   std::shared_ptr<Buffer> 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));
 

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)

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

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):

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 = <int8_t> CFileType_NonExistent
+    Unknown = <int8_t> CFileType_Unknown
+    File = <int8_t> CFileType_File
+    Directory = <int8_t> 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)