Add more filesystem methods, tests for HDFS

Change-Id: I6ca7a89c0cd9219657c4964345ff7664b8c3dd78

Author: wesm

cpp/src/arrow/io/hdfs.cc

@@ -495,7 +495,7 @@ class HadoopFileSystem::HadoopFileSystemImpl {
   }
 
   Status Chmod(const std::string& path, int mode) {
-    int ret = driver_->Chmod(fs_, path.c_str(), static_cast<short>(mode));
+    int ret = driver_->Chmod(fs_, path.c_str(), static_cast<short>(mode));  // NOLINT
     CHECK_FAILURE(ret, "Chmod");
     return Status::OK();
   }

python/doc/source/api.rst

@@ -165,14 +165,16 @@ Input / Output and Shared Memory
    PythonFile
 
 File Systems
------------
+------------
 
 .. autosummary::
    :toctree: generated/
 
    hdfs.connect
-   HadoopFileSystem
-   LocalFilesystem
+   LocalFileSystem
+
+.. class:: HadoopFileSystem
+   :noindex:
 
 .. _api.ipc:
 

python/doc/source/conf.py

@@ -62,7 +62,8 @@
 ]
 
 # Show members for classes in .. autosummary
-autodoc_default_flags = ['members', 'undoc-members', 'show-inheritance', 'inherited-members']
+autodoc_default_flags = ['members', 'undoc-members', 'show-inheritance',
+                         'inherited-members']
 
 # numpydoc configuration
 napoleon_use_rtype = False

python/doc/source/filesystems.rst

@@ -15,8 +15,8 @@
 .. specific language governing permissions and limitations
 .. under the License.
 
-Filesystem Interfaces
-=====================
+File System Interfaces
+======================
 
 In this section, we discuss filesystem-like interfaces in PyArrow.
 
@@ -34,7 +34,7 @@ System. You connect like so:
    hdfs = pa.hdfs.connect(host, port, user=user, kerb_ticket=ticket_cache_path)
    type(hdfs)
 
-By default, ``pyarrow.hdfs.HadoopFilesystem`` uses libhdfs, a JNI-based
+By default, ``pyarrow.hdfs.HadoopFileSystem`` uses libhdfs, a JNI-based
 interface to the Java Hadoop client. This library is loaded **at runtime**
 (rather than at link / library load time, since the library may not be in your
 LD_LIBRARY_PATH), and relies on some environment variables.
@@ -69,16 +69,20 @@ HDFS API
    :toctree: generated/
 
    hdfs.connect
-   HadoopFilesystem
-   HadoopFilesystem.cat
-   HadoopFilesystem.chmod
-   HadoopFilesystem.chown
-   HadoopFilesystem.delete
-   HadoopFilesystem.download
-   HadoopFilesystem.exists
-   HadoopFilesystem.info
-   HadoopFilesystem.ls
-   HadoopFilesystem.mkdir
-   HadoopFilesystem.rm
-   HadoopFilesystem.upload
+   HadoopFileSystem.cat
+   HadoopFileSystem.chmod
+   HadoopFileSystem.chown
+   HadoopFileSystem.delete
+   HadoopFileSystem.df
+   HadoopFileSystem.disk_usage
+   HadoopFileSystem.download
+   HadoopFileSystem.exists
+   HadoopFileSystem.get_capacity
+   HadoopFileSystem.get_space_used
+   HadoopFileSystem.info
+   HadoopFileSystem.ls
+   HadoopFileSystem.mkdir
+   HadoopFileSystem.rename
+   HadoopFileSystem.rm
+   HadoopFileSystem.upload
    HdfsFile

python/doc/source/memory.rst

@@ -226,10 +226,3 @@ file interfaces that can read and write to Arrow Buffers.
    reader.read(7)
 
 These have similar semantics to Python's built-in ``io.BytesIO``.
-
-Hadoop Filesystem
------------------
-
-:class:`~pyarrow.HdfsFile` is an implementation of :class:`~pyarrow.NativeFile`
-that can read and write to the Hadoop filesytem. Read more in the
-:ref:`Filesystems Section <hdfs>`.

python/pyarrow/filesystem.py

@@ -26,6 +26,17 @@ class FileSystem(object):
     """
     Abstract filesystem interface
     """
+    def cat(self, path):
+        """
+        Return contents of file as a bytes object
+
+        Returns
+        -------
+        contents : bytes
+        """
+        with self.open(path, 'rb') as f:
+            return f.read()
+
     def ls(self, path):
         """
         Return list of file paths
@@ -44,12 +55,68 @@ def delete(self, path, recursive=False):
         """
         raise NotImplementedError
 
+    def disk_usage(self, path):
+        """
+        Compute bytes used by all contents under indicated path in file tree
+
+        Parameters
+        ----------
+        path : string
+            Can be a file path or directory
+
+        Returns
+        -------
+        usage : int
+        """
+        path_info = self.stat(path)
+        if path_info['kind'] == 'file':
+            return path_info['size']
+
+        total = 0
+        for root, directories, files in self.walk(path):
+            for child_path in files:
+                abspath = self._path_join(root, child_path)
+                total += self.stat(abspath)['size']
+
+        return total
+
+    def _path_join(self, *args):
+        return self.pathsep.join(args)
+
+    def stat(self, path):
+        """
+
+        Returns
+        -------
+        stat : dict
+        """
+        raise NotImplementedError('FileSystem.stat')
+
     def rm(self, path, recursive=False):
         """
         Alias for FileSystem.delete
         """
         return self.delete(path, recursive=recursive)
 
+    def mv(self, path, new_path):
+        """
+        Alias for FileSystem.rename
+        """
+        return self.rename(path, new_path)
+
+    def rename(self, path, new_path):
+        """
+        Rename file, like UNIX mv command
+
+        Parameters
+        ----------
+        path : string
+            Path to alter
+        new_path : string
+            Path to move to
+        """
+        raise NotImplementedError('FileSystem.rename')
+
     def mkdir(self, path, create_parents=True):
         raise NotImplementedError
 

python/pyarrow/hdfs.py

@@ -48,6 +48,10 @@ def delete(self, path, recursive=False):
     def mkdir(self, path, create_parents=True):
         return super(HadoopFileSystem, self).mkdir(path)
 
+    @implements(FileSystem.rename)
+    def rename(self, path, new_path):
+        return super(HadoopFileSystem, self).rename(path, new_path)
+
     def ls(self, path, detail=False):
         """
         Retrieve directory contents and metadata, if requested.
@@ -82,7 +86,7 @@ def walk(self, top_path):
         directories, files = _libhdfs_walk_files_dirs(top_path, contents)
         yield top_path, directories, files
         for dirname in directories:
-            for tup in self.walk(dirname):
+            for tup in self.walk(self._path_join(top_path, dirname)):
                 yield tup
 
 

python/pyarrow/includes/libarrow.pxd

@@ -413,6 +413,10 @@ cdef extern from "arrow/io/interfaces.h" namespace "arrow::io" nogil:
         ObjectType_FILE" arrow::io::ObjectType::FILE"
         ObjectType_DIRECTORY" arrow::io::ObjectType::DIRECTORY"
 
+    cdef cppclass FileStatistics:
+        int64_t size
+        ObjectType kind
+
     cdef cppclass FileInterface:
         CStatus Close()
         CStatus Tell(int64_t* position)
@@ -450,6 +454,9 @@ cdef extern from "arrow/io/interfaces.h" namespace "arrow::io" nogil:
                                          WriteableFile):
         pass
 
+    cdef cppclass FileSystem:
+        CStatus Stat(const c_string& path, FileStatistics* stat)
+
 
 cdef extern from "arrow/io/file.h" namespace "arrow::io" nogil:
 
@@ -517,7 +524,7 @@ cdef extern from "arrow/io/hdfs.h" namespace "arrow::io" nogil:
     cdef cppclass HdfsOutputStream(OutputStream):
         pass
 
-    cdef cppclass CHadoopFileSystem" arrow::io::HadoopFileSystem":
+    cdef cppclass CHadoopFileSystem" arrow::io::HadoopFileSystem"(FileSystem):
         @staticmethod
         CStatus Connect(const HdfsConnectionConfig* config,
                         shared_ptr[CHadoopFileSystem]* client)

python/pyarrow/io-hdfs.pxi

@@ -129,6 +129,48 @@ cdef class HadoopFileSystem:
         self._path_info(path, &info)
         return info.kind == ObjectType_FILE
 
+    def get_capacity(self):
+        """
+        Get reported total capacity of file system
+
+        Returns
+        -------
+        capacity : int
+        """
+        cdef int64_t capacity = 0
+        with nogil:
+            check_status(self.client.get().GetCapacity(&capacity))
+        return capacity
+
+    def get_space_used(self):
+        """
+        Get space used on file system
+
+        Returns
+        -------
+        space_used : int
+        """
+        cdef int64_t space_used = 0
+        with nogil:
+            check_status(self.client.get().GetUsed(&space_used))
+        return space_used
+
+    def df(self):
+        """
+        Return free space on disk, like the UNIX df command
+
+        Returns
+        -------
+        space : int
+        """
+        return self.get_capacity() - self.get_space_used()
+
+    def rename(self, path, new_path):
+        cdef c_string c_path = tobytes(path)
+        cdef c_string c_new_path = tobytes(new_path)
+        with nogil:
+            check_status(self.client.get().Rename(c_path, c_new_path))
+
     def info(self, path):
         """
         Return detailed HDFS information for path
@@ -158,6 +200,30 @@ cdef class HadoopFileSystem:
                      else 'file')
         }
 
+    def stat(self, path):
+        """
+        Return basic file system statistics about path
+
+        Parameters
+        ----------
+        path : string
+            Path to file or directory
+
+        Returns
+        -------
+        stat : dict
+        """
+        cdef FileStatistics info
+        cdef c_string c_path = tobytes(path)
+        with nogil:
+            check_status(self.client.get()
+                         .Stat(c_path, &info))
+        return {
+            'size': info.size,
+            'kind': ('directory' if info.kind == ObjectType_DIRECTORY
+                     else 'file')
+        }
+
     cdef _path_info(self, path, HdfsPathInfo* info):
         cdef c_string c_path = tobytes(path)
 
@@ -290,9 +356,16 @@ cdef class HadoopFileSystem:
     def open(self, path, mode='rb', buffer_size=None, replication=None,
              default_block_size=None):
         """
+        Open HDFS file for reading or writing
+
         Parameters
         ----------
-        mode : string, 'rb', 'wb', 'ab'
+        mode : string
+            Must be one of 'rb', 'wb', 'ab'
+
+        Returns
+        -------
+        handle : HdfsFile
         """
         self._ensure_client()