From b4c86e0c05b43fc8554d2f3b407f8d7c36b716d4 Mon Sep 17 00:00:00 2001
From: Joris Van den Bossche <jorisvandenbossche@gmail.com>
Date: Mon, 13 Sep 2021 16:03:41 +0300
Subject: [PATCH 01/21] arrow: implement arrow wrapper

---
 fsspec/implementations/arrow.py | 273 ++++++++++++++++++++++++++++++++
 1 file changed, 273 insertions(+)
 create mode 100644 fsspec/implementations/arrow.py

diff --git a/fsspec/implementations/arrow.py b/fsspec/implementations/arrow.py
new file mode 100644
index 000000000..c01e10892
--- /dev/null
+++ b/fsspec/implementations/arrow.py
@@ -0,0 +1,273 @@
+import datetime
+import io
+import os
+import posixpath
+import re
+import shutil
+import tempfile
+
+from fsspec import AbstractFileSystem
+from fsspec.utils import stringify_path
+
+
+class ArrowFSWrapper(AbstractFileSystem):
+    """FSSpec-compatible wrapper of pyarrow.fs.FileSystem.
+
+    Parameters
+    ----------
+    fs : pyarrow.fs.FileSystem
+
+    """
+
+    # root_marker = "/"
+    # protocol = "file"
+    # local_file = True
+
+    def __init__(self, fs, **kwargs):
+        super().__init__(**kwargs)
+        try:
+            import pyarrow.fs
+        except ImportError:
+            raise ImportError("pyarrow required to use the ArrowFSWrapper")
+
+        if not isinstance(fs, pyarrow.fs.FileSystem):
+            raise TypeError(
+                "'fs' should be an instance of a pyarrow.fs.FileSystem subclass"
+            )
+        self.fs = fs
+
+    def mkdir(self, path, create_parents=True, **kwargs):
+        path = self._strip_protocol(path)
+        if create_parents:
+            self.makedirs(path, exist_ok=True)
+        else:
+            self.fs.create_dir(path, recursive=False)
+
+    def makedirs(self, path, exist_ok=False):
+        path = self._strip_protocol(path)
+        self.fs.create_dir(path, recursive=True)
+
+    def rmdir(self, path):
+        path = self._strip_protocol(path)
+        self.fs.delete_dir(path)
+
+    def ls(self, path, detail=False, **kwargs):
+        path = self._strip_protocol(path)
+        paths = [posixpath.join(path, f) for f in os.listdir(path)]
+        if detail:
+            return [self.info(f) for f in paths]
+        else:
+            return paths
+
+    def glob(self, path, **kwargs):
+        path = self._strip_protocol(path)
+        return super().glob(path, **kwargs)
+
+    def info(self, path, **kwargs):
+        print("getting info for ", path)
+        path = self._strip_protocol(path)
+        info = self.fs.get_file_info([path])[0]
+        dest = False
+
+        from pyarrow.fs import FileType
+
+        if info.type == FileType.Directory:
+            t = "directory"
+        elif info.is_file:
+            t = "file"
+        else:
+            t = "other"
+        result = {
+            "name": path,
+            "size": info.size,
+            "type": t,
+            "created": None,
+            "mtime": info.mtime,
+        }
+        return result
+
+    def cp_file(self, path1, path2, **kwargs):
+        path1 = self._strip_protocol(path1).rstrip("/")
+        path2 = self._strip_protocol(path2).rstrip("/")
+        if self.auto_mkdir:
+            self.makedirs(self._parent(path2), exist_ok=True)
+        if self.isfile(path1):
+            shutil.copyfile(path1, path2)
+        else:
+            self.mkdirs(path2, exist_ok=True)
+
+    def get_file(self, path1, path2, **kwargs):
+        return self.cp_file(path1, path2, **kwargs)
+
+    def put_file(self, path1, path2, **kwargs):
+        return self.cp_file(path1, path2, **kwargs)
+
+    def mv_file(self, path1, path2, **kwargs):
+        path1 = self._strip_protocol(path1).rstrip("/")
+        path2 = self._strip_protocol(path2).rstrip("/")
+        os.rename(path1, path2)
+
+    def rm(self, path, recursive=False, maxdepth=None):
+        path = self._strip_protocol(path).rstrip("/")
+        if recursive and self.isdir(path):
+            self.fs.delete_dir(path)
+        else:
+            self.fs.delete_file(path)
+
+    def _open(self, path, mode="rb", block_size=None, **kwargs):
+        path = self._strip_protocol(path)
+        if self.auto_mkdir and "w" in mode:
+            self.makedirs(self._parent(path), exist_ok=True)
+        return LocalFileOpener(path, mode, fs=self, **kwargs)
+
+    def touch(self, path, **kwargs):
+        path = self._strip_protocol(path)
+        # if self.auto_mkdir:
+        #     self.makedirs(self._parent(path), exist_ok=True)
+        if self.exists(path):
+            pass  # os.utime(path, None)
+        else:
+            self.fs.open_input_file(path).close()
+
+    def created(self, path):
+        info = self.info(path=path)
+        return datetime.datetime.utcfromtimestamp(info["created"])
+
+    def modified(self, path):
+        info = self.info(path=path)
+        return datetime.datetime.utcfromtimestamp(info["mtime"])
+
+    @classmethod
+    def _parent(cls, path):
+        path = cls._strip_protocol(path).rstrip("/")
+        if "/" in path:
+            return path.rsplit("/", 1)[0]
+        else:
+            return cls.root_marker
+
+    @classmethod
+    def _strip_protocol(cls, path):
+        path = stringify_path(path)
+        if path.startswith("file://"):
+            path = path[7:]
+        path = os.path.expanduser(path)
+        return make_path_posix(path)
+
+    def _isfilestore(self):
+        # Inheriting from DaskFileSystem makes this False (S3, etc. were)
+        # the original motivation. But we are a posix-like file system.
+        # See https://github.com/dask/dask/issues/5526
+        return True
+
+
+def make_path_posix(path, sep=os.sep):
+    """ Make path generic """
+    if isinstance(path, (list, set, tuple)):
+        return type(path)(make_path_posix(p) for p in path)
+    if re.match("/[A-Za-z]:", path):
+        # for windows file URI like "file:///C:/folder/file"
+        # or "file:///C:\\dir\\file"
+        path = path[1:]
+    if path.startswith("\\\\"):
+        # special case for windows UNC/DFS-style paths, do nothing,
+        # just flip the slashes around (case below does not work!)
+        return path.replace("\\", "/")
+    if re.match("[A-Za-z]:", path):
+        # windows full path like "C:\\local\\path"
+        return path.lstrip("\\").replace("\\", "/").replace("//", "/")
+    if path.startswith("\\"):
+        # windows network path like "\\server\\path"
+        return "/" + path.lstrip("\\").replace("\\", "/").replace("//", "/")
+    if (
+        sep not in path
+        and "/" not in path
+        or (sep == "/" and not path.startswith("/"))
+        or (sep == "\\" and ":" not in path)
+    ):
+        # relative path like "path" or "rel\\path" (win) or rel/path"
+        path = os.path.abspath(path)
+        if os.sep == "\\":
+            # abspath made some more '\\' separators
+            return make_path_posix(path, sep)
+    return path
+
+
+class LocalFileOpener(object):
+    def __init__(self, path, mode, autocommit=True, fs=None, **kwargs):
+        self.path = path
+        self.mode = mode
+        self.fs = fs
+        self.f = None
+        self.autocommit = autocommit
+        self.blocksize = io.DEFAULT_BUFFER_SIZE
+        self._open()
+
+    def _open(self):
+        if self.f is None or self.f.closed:
+            if self.autocommit or "w" not in self.mode:
+                self.f = open(self.path, mode=self.mode)
+            else:
+                # TODO: check if path is writable?
+                i, name = tempfile.mkstemp()
+                os.close(i)  # we want normal open and normal buffered file
+                self.temp = name
+                self.f = open(name, mode=self.mode)
+            if "w" not in self.mode:
+                self.details = self.fs.info(self.path)
+                self.size = self.details["size"]
+                self.f.size = self.size
+
+    def _fetch_range(self, start, end):
+        # probably only used by cached FS
+        if "r" not in self.mode:
+            raise ValueError
+        self._open()
+        self.f.seek(start)
+        return self.f.read(end - start)
+
+    def __setstate__(self, state):
+        self.f = None
+        loc = state.pop("loc", None)
+        self.__dict__.update(state)
+        if "r" in state["mode"]:
+            self.f = None
+            self._open()
+            self.f.seek(loc)
+
+    def __getstate__(self):
+        d = self.__dict__.copy()
+        d.pop("f")
+        if "r" in self.mode:
+            d["loc"] = self.f.tell()
+        else:
+            if not self.f.closed:
+                raise ValueError("Cannot serialise open write-mode local file")
+        return d
+
+    def commit(self):
+        if self.autocommit:
+            raise RuntimeError("Can only commit if not already set to autocommit")
+        os.replace(self.temp, self.path)
+
+    def discard(self):
+        if self.autocommit:
+            raise RuntimeError("Cannot discard if set to autocommit")
+        os.remove(self.temp)
+
+    def __fspath__(self):
+        # uniquely among fsspec implementations, this is a real, local path
+        return self.path
+
+    def __iter__(self):
+        return self.f.__iter__()
+
+    def __getattr__(self, item):
+        return getattr(self.f, item)
+
+    def __enter__(self):
+        self._incontext = True
+        return self.f.__enter__()
+
+    def __exit__(self, exc_type, exc_value, traceback):
+        self._incontext = False
+        self.f.__exit__(exc_type, exc_value, traceback)

From 1131c8c12ad2abf2deb13ee9e5ac668d17008fed Mon Sep 17 00:00:00 2001
From: Batuhan Taskaya <batuhan@python.org>
Date: Mon, 13 Sep 2021 16:06:37 +0300
Subject: [PATCH 02/21] arrow: share make_path_posix with localfs

---
 fsspec/implementations/arrow.py | 33 +--------------------------------
 1 file changed, 1 insertion(+), 32 deletions(-)

diff --git a/fsspec/implementations/arrow.py b/fsspec/implementations/arrow.py
index c01e10892..dff30f243 100644
--- a/fsspec/implementations/arrow.py
+++ b/fsspec/implementations/arrow.py
@@ -7,6 +7,7 @@
 import tempfile
 
 from fsspec import AbstractFileSystem
+from fsspec.implementations.local import make_path_posix
 from fsspec.utils import stringify_path
 
 
@@ -160,38 +161,6 @@ def _isfilestore(self):
         return True
 
 
-def make_path_posix(path, sep=os.sep):
-    """ Make path generic """
-    if isinstance(path, (list, set, tuple)):
-        return type(path)(make_path_posix(p) for p in path)
-    if re.match("/[A-Za-z]:", path):
-        # for windows file URI like "file:///C:/folder/file"
-        # or "file:///C:\\dir\\file"
-        path = path[1:]
-    if path.startswith("\\\\"):
-        # special case for windows UNC/DFS-style paths, do nothing,
-        # just flip the slashes around (case below does not work!)
-        return path.replace("\\", "/")
-    if re.match("[A-Za-z]:", path):
-        # windows full path like "C:\\local\\path"
-        return path.lstrip("\\").replace("\\", "/").replace("//", "/")
-    if path.startswith("\\"):
-        # windows network path like "\\server\\path"
-        return "/" + path.lstrip("\\").replace("\\", "/").replace("//", "/")
-    if (
-        sep not in path
-        and "/" not in path
-        or (sep == "/" and not path.startswith("/"))
-        or (sep == "\\" and ":" not in path)
-    ):
-        # relative path like "path" or "rel\\path" (win) or rel/path"
-        path = os.path.abspath(path)
-        if os.sep == "\\":
-            # abspath made some more '\\' separators
-            return make_path_posix(path, sep)
-    return path
-
-
 class LocalFileOpener(object):
     def __init__(self, path, mode, autocommit=True, fs=None, **kwargs):
         self.path = path

From 4aa8dbcfe400927e4b0ced396f8c7524b62fb88c Mon Sep 17 00:00:00 2001
From: Batuhan Taskaya <batuhan@python.org>
Date: Mon, 13 Sep 2021 16:11:36 +0300
Subject: [PATCH 03/21] arrow: get rid of LocalFileOpener, migrate to
 fs.open_*_stream

---
 fsspec/implementations/arrow.py | 95 ++++-----------------------------
 1 file changed, 10 insertions(+), 85 deletions(-)

diff --git a/fsspec/implementations/arrow.py b/fsspec/implementations/arrow.py
index dff30f243..9a760c0c0 100644
--- a/fsspec/implementations/arrow.py
+++ b/fsspec/implementations/arrow.py
@@ -6,6 +6,8 @@
 import shutil
 import tempfile
 
+from contextlib import closing
+
 from fsspec import AbstractFileSystem
 from fsspec.implementations.local import make_path_posix
 from fsspec.utils import stringify_path
@@ -116,10 +118,14 @@ def rm(self, path, recursive=False, maxdepth=None):
             self.fs.delete_file(path)
 
     def _open(self, path, mode="rb", block_size=None, **kwargs):
-        path = self._strip_protocol(path)
-        if self.auto_mkdir and "w" in mode:
-            self.makedirs(self._parent(path), exist_ok=True)
-        return LocalFileOpener(path, mode, fs=self, **kwargs)
+        if mode == "rb":
+            stream = self.fs.open_input_stream(path)
+        elif mode == "wb":
+            stream = self.fs.open_output_stream(path)
+        else:
+            raise ValueError(f"unsupported mode for Arrow filesystem: {mode!r}")
+
+        return closing(stream)
 
     def touch(self, path, **kwargs):
         path = self._strip_protocol(path)
@@ -159,84 +165,3 @@ def _isfilestore(self):
         # the original motivation. But we are a posix-like file system.
         # See https://github.com/dask/dask/issues/5526
         return True
-
-
-class LocalFileOpener(object):
-    def __init__(self, path, mode, autocommit=True, fs=None, **kwargs):
-        self.path = path
-        self.mode = mode
-        self.fs = fs
-        self.f = None
-        self.autocommit = autocommit
-        self.blocksize = io.DEFAULT_BUFFER_SIZE
-        self._open()
-
-    def _open(self):
-        if self.f is None or self.f.closed:
-            if self.autocommit or "w" not in self.mode:
-                self.f = open(self.path, mode=self.mode)
-            else:
-                # TODO: check if path is writable?
-                i, name = tempfile.mkstemp()
-                os.close(i)  # we want normal open and normal buffered file
-                self.temp = name
-                self.f = open(name, mode=self.mode)
-            if "w" not in self.mode:
-                self.details = self.fs.info(self.path)
-                self.size = self.details["size"]
-                self.f.size = self.size
-
-    def _fetch_range(self, start, end):
-        # probably only used by cached FS
-        if "r" not in self.mode:
-            raise ValueError
-        self._open()
-        self.f.seek(start)
-        return self.f.read(end - start)
-
-    def __setstate__(self, state):
-        self.f = None
-        loc = state.pop("loc", None)
-        self.__dict__.update(state)
-        if "r" in state["mode"]:
-            self.f = None
-            self._open()
-            self.f.seek(loc)
-
-    def __getstate__(self):
-        d = self.__dict__.copy()
-        d.pop("f")
-        if "r" in self.mode:
-            d["loc"] = self.f.tell()
-        else:
-            if not self.f.closed:
-                raise ValueError("Cannot serialise open write-mode local file")
-        return d
-
-    def commit(self):
-        if self.autocommit:
-            raise RuntimeError("Can only commit if not already set to autocommit")
-        os.replace(self.temp, self.path)
-
-    def discard(self):
-        if self.autocommit:
-            raise RuntimeError("Cannot discard if set to autocommit")
-        os.remove(self.temp)
-
-    def __fspath__(self):
-        # uniquely among fsspec implementations, this is a real, local path
-        return self.path
-
-    def __iter__(self):
-        return self.f.__iter__()
-
-    def __getattr__(self, item):
-        return getattr(self.f, item)
-
-    def __enter__(self):
-        self._incontext = True
-        return self.f.__enter__()
-
-    def __exit__(self, exc_type, exc_value, traceback):
-        self._incontext = False
-        self.f.__exit__(exc_type, exc_value, traceback)

From a0a37f87ccaf195e0c8d194682d188157bb73091 Mon Sep 17 00:00:00 2001
From: Batuhan Taskaya <batuhan@python.org>
Date: Mon, 13 Sep 2021 16:12:53 +0300
Subject: [PATCH 04/21] arrow: remove debug statements

---
 fsspec/implementations/arrow.py | 2 --
 1 file changed, 2 deletions(-)

diff --git a/fsspec/implementations/arrow.py b/fsspec/implementations/arrow.py
index 9a760c0c0..77049c904 100644
--- a/fsspec/implementations/arrow.py
+++ b/fsspec/implementations/arrow.py
@@ -5,7 +5,6 @@
 import re
 import shutil
 import tempfile
-
 from contextlib import closing
 
 from fsspec import AbstractFileSystem
@@ -67,7 +66,6 @@ def glob(self, path, **kwargs):
         return super().glob(path, **kwargs)
 
     def info(self, path, **kwargs):
-        print("getting info for ", path)
         path = self._strip_protocol(path)
         info = self.fs.get_file_info([path])[0]
         dest = False

From c1ac73dc927f6b6e53314cabb18bace9d2bb805e Mon Sep 17 00:00:00 2001
From: Batuhan Taskaya <batuhan@python.org>
Date: Mon, 13 Sep 2021 16:19:15 +0300
Subject: [PATCH 05/21] arrow: cleanup info()

---
 fsspec/implementations/arrow.py | 30 ++++++++++++++----------------
 1 file changed, 14 insertions(+), 16 deletions(-)

diff --git a/fsspec/implementations/arrow.py b/fsspec/implementations/arrow.py
index 77049c904..5b3181d1b 100644
--- a/fsspec/implementations/arrow.py
+++ b/fsspec/implementations/arrow.py
@@ -61,31 +61,29 @@ def ls(self, path, detail=False, **kwargs):
         else:
             return paths
 
-    def glob(self, path, **kwargs):
-        path = self._strip_protocol(path)
-        return super().glob(path, **kwargs)
-
     def info(self, path, **kwargs):
+        from pyarrow.fs import FileType
+
         path = self._strip_protocol(path)
-        info = self.fs.get_file_info([path])[0]
-        dest = False
+        [info] = self.fs.get_file_info([path])
+        return self._make_entry(info)
 
+    def _make_entry(self, info):
         from pyarrow.fs import FileType
 
-        if info.type == FileType.Directory:
-            t = "directory"
-        elif info.is_file:
-            t = "file"
+        if info.type is FileType.Directory:
+            kind = "directory"
+        elif info.type is FileType.File:
+            kind = "file"
         else:
-            t = "other"
-        result = {
-            "name": path,
+            kind = "other"
+
+        return {
+            "name": info.path,
             "size": info.size,
-            "type": t,
-            "created": None,
+            "type": kind,
             "mtime": info.mtime,
         }
-        return result
 
     def cp_file(self, path1, path2, **kwargs):
         path1 = self._strip_protocol(path1).rstrip("/")

From a0340e50793049d2e2757a74fdce0dcaed99b81f Mon Sep 17 00:00:00 2001
From: Batuhan Taskaya <batuhan@python.org>
Date: Mon, 13 Sep 2021 16:19:40 +0300
Subject: [PATCH 06/21] arrow: more cleanups to satisfy linters

---
 fsspec/implementations/arrow.py | 5 -----
 1 file changed, 5 deletions(-)

diff --git a/fsspec/implementations/arrow.py b/fsspec/implementations/arrow.py
index 5b3181d1b..5c7dbc6be 100644
--- a/fsspec/implementations/arrow.py
+++ b/fsspec/implementations/arrow.py
@@ -1,10 +1,7 @@
 import datetime
-import io
 import os
 import posixpath
-import re
 import shutil
-import tempfile
 from contextlib import closing
 
 from fsspec import AbstractFileSystem
@@ -62,8 +59,6 @@ def ls(self, path, detail=False, **kwargs):
             return paths
 
     def info(self, path, **kwargs):
-        from pyarrow.fs import FileType
-
         path = self._strip_protocol(path)
         [info] = self.fs.get_file_info([path])
         return self._make_entry(info)

From 7b0e3f2dce56ab6685e68a2ad72424c9dea9c7d1 Mon Sep 17 00:00:00 2001
From: Batuhan Taskaya <batuhan@python.org>
Date: Mon, 13 Sep 2021 16:23:40 +0300
Subject: [PATCH 07/21] arrow: implement rm_file, get rid of some methods

---
 fsspec/implementations/arrow.py | 75 ++++++++++++---------------------
 1 file changed, 28 insertions(+), 47 deletions(-)

diff --git a/fsspec/implementations/arrow.py b/fsspec/implementations/arrow.py
index 5c7dbc6be..a9c897f2e 100644
--- a/fsspec/implementations/arrow.py
+++ b/fsspec/implementations/arrow.py
@@ -1,4 +1,3 @@
-import datetime
 import os
 import posixpath
 import shutil
@@ -35,20 +34,21 @@ def __init__(self, fs, **kwargs):
             )
         self.fs = fs
 
-    def mkdir(self, path, create_parents=True, **kwargs):
-        path = self._strip_protocol(path)
-        if create_parents:
-            self.makedirs(path, exist_ok=True)
+    @classmethod
+    def _parent(cls, path):
+        path = cls._strip_protocol(path).rstrip("/")
+        if "/" in path:
+            return path.rsplit("/", 1)[0]
         else:
-            self.fs.create_dir(path, recursive=False)
-
-    def makedirs(self, path, exist_ok=False):
-        path = self._strip_protocol(path)
-        self.fs.create_dir(path, recursive=True)
+            return cls.root_marker
 
-    def rmdir(self, path):
-        path = self._strip_protocol(path)
-        self.fs.delete_dir(path)
+    @classmethod
+    def _strip_protocol(cls, path):
+        path = stringify_path(path)
+        if path.startswith("file://"):
+            path = path[7:]
+        path = os.path.expanduser(path)
+        return make_path_posix(path)
 
     def ls(self, path, detail=False, **kwargs):
         path = self._strip_protocol(path)
@@ -85,6 +85,7 @@ def cp_file(self, path1, path2, **kwargs):
         path2 = self._strip_protocol(path2).rstrip("/")
         if self.auto_mkdir:
             self.makedirs(self._parent(path2), exist_ok=True)
+
         if self.isfile(path1):
             shutil.copyfile(path1, path2)
         else:
@@ -101,6 +102,10 @@ def mv_file(self, path1, path2, **kwargs):
         path2 = self._strip_protocol(path2).rstrip("/")
         os.rename(path1, path2)
 
+    def rm_file(self, path):
+        path = self._strip_protocol(path)
+        self.fs.delete_file(path)
+
     def rm(self, path, recursive=False, maxdepth=None):
         path = self._strip_protocol(path).rstrip("/")
         if recursive and self.isdir(path):
@@ -118,41 +123,17 @@ def _open(self, path, mode="rb", block_size=None, **kwargs):
 
         return closing(stream)
 
-    def touch(self, path, **kwargs):
+    def mkdir(self, path, create_parents=True, **kwargs):
         path = self._strip_protocol(path)
-        # if self.auto_mkdir:
-        #     self.makedirs(self._parent(path), exist_ok=True)
-        if self.exists(path):
-            pass  # os.utime(path, None)
-        else:
-            self.fs.open_input_file(path).close()
-
-    def created(self, path):
-        info = self.info(path=path)
-        return datetime.datetime.utcfromtimestamp(info["created"])
-
-    def modified(self, path):
-        info = self.info(path=path)
-        return datetime.datetime.utcfromtimestamp(info["mtime"])
-
-    @classmethod
-    def _parent(cls, path):
-        path = cls._strip_protocol(path).rstrip("/")
-        if "/" in path:
-            return path.rsplit("/", 1)[0]
+        if create_parents:
+            self.makedirs(path, exist_ok=True)
         else:
-            return cls.root_marker
+            self.fs.create_dir(path, recursive=False)
 
-    @classmethod
-    def _strip_protocol(cls, path):
-        path = stringify_path(path)
-        if path.startswith("file://"):
-            path = path[7:]
-        path = os.path.expanduser(path)
-        return make_path_posix(path)
+    def makedirs(self, path, exist_ok=False):
+        path = self._strip_protocol(path)
+        self.fs.create_dir(path, recursive=True)
 
-    def _isfilestore(self):
-        # Inheriting from DaskFileSystem makes this False (S3, etc. were)
-        # the original motivation. But we are a posix-like file system.
-        # See https://github.com/dask/dask/issues/5526
-        return True
+    def rmdir(self, path):
+        path = self._strip_protocol(path)
+        self.fs.delete_dir(path)

From 7d5149f85df2efa9e87712a8e27e20fc7798b4a3 Mon Sep 17 00:00:00 2001
From: Batuhan Taskaya <batuhan@python.org>
Date: Mon, 13 Sep 2021 16:24:43 +0300
Subject: [PATCH 08/21] arrow: use self.fs.move instead of os.rename

---
 fsspec/implementations/arrow.py | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/fsspec/implementations/arrow.py b/fsspec/implementations/arrow.py
index a9c897f2e..3cbbb6c22 100644
--- a/fsspec/implementations/arrow.py
+++ b/fsspec/implementations/arrow.py
@@ -100,7 +100,7 @@ def put_file(self, path1, path2, **kwargs):
     def mv_file(self, path1, path2, **kwargs):
         path1 = self._strip_protocol(path1).rstrip("/")
         path2 = self._strip_protocol(path2).rstrip("/")
-        os.rename(path1, path2)
+        self.fs.move(path1, path2)
 
     def rm_file(self, path):
         path = self._strip_protocol(path)

From 3e1878df91c782e5f54f26a31f5cbb377cbbb391 Mon Sep 17 00:00:00 2001
From: Batuhan Taskaya <batuhan@python.org>
Date: Mon, 13 Sep 2021 16:26:19 +0300
Subject: [PATCH 09/21] fs: use fs.copy_file instead of shutil.copyfile

---
 fsspec/implementations/arrow.py | 6 +-----
 1 file changed, 1 insertion(+), 5 deletions(-)

diff --git a/fsspec/implementations/arrow.py b/fsspec/implementations/arrow.py
index 3cbbb6c22..9b66a63d6 100644
--- a/fsspec/implementations/arrow.py
+++ b/fsspec/implementations/arrow.py
@@ -1,6 +1,5 @@
 import os
 import posixpath
-import shutil
 from contextlib import closing
 
 from fsspec import AbstractFileSystem
@@ -86,10 +85,7 @@ def cp_file(self, path1, path2, **kwargs):
         if self.auto_mkdir:
             self.makedirs(self._parent(path2), exist_ok=True)
 
-        if self.isfile(path1):
-            shutil.copyfile(path1, path2)
-        else:
-            self.mkdirs(path2, exist_ok=True)
+        self.fs.copy_file(path1, path2)
 
     def get_file(self, path1, path2, **kwargs):
         return self.cp_file(path1, path2, **kwargs)

From 918bd25490ec8fc6c9c7742ade896e4ba713b30c Mon Sep 17 00:00:00 2001
From: Batuhan Taskaya <batuhan@python.org>
Date: Mon, 13 Sep 2021 16:26:41 +0300
Subject: [PATCH 10/21] arrow: remove auto_mkdir related logic since this is
 not localfs

---
 fsspec/implementations/arrow.py | 3 ---
 1 file changed, 3 deletions(-)

diff --git a/fsspec/implementations/arrow.py b/fsspec/implementations/arrow.py
index 9b66a63d6..a34838f1b 100644
--- a/fsspec/implementations/arrow.py
+++ b/fsspec/implementations/arrow.py
@@ -82,9 +82,6 @@ def _make_entry(self, info):
     def cp_file(self, path1, path2, **kwargs):
         path1 = self._strip_protocol(path1).rstrip("/")
         path2 = self._strip_protocol(path2).rstrip("/")
-        if self.auto_mkdir:
-            self.makedirs(self._parent(path2), exist_ok=True)
-
         self.fs.copy_file(path1, path2)
 
     def get_file(self, path1, path2, **kwargs):

From b7ce60121805dad0de8bd0eb396fb42c887d382a Mon Sep 17 00:00:00 2001
From: Batuhan Taskaya <batuhan@python.org>
Date: Mon, 13 Sep 2021 16:33:33 +0300
Subject: [PATCH 11/21] arrow: support different schemes as well

---
 fsspec/implementations/arrow.py | 20 +++++++++-----------
 1 file changed, 9 insertions(+), 11 deletions(-)

diff --git a/fsspec/implementations/arrow.py b/fsspec/implementations/arrow.py
index a34838f1b..74d887f94 100644
--- a/fsspec/implementations/arrow.py
+++ b/fsspec/implementations/arrow.py
@@ -1,9 +1,6 @@
-import os
-import posixpath
 from contextlib import closing
 
 from fsspec import AbstractFileSystem
-from fsspec.implementations.local import make_path_posix
 from fsspec.utils import stringify_path
 
 
@@ -44,18 +41,19 @@ def _parent(cls, path):
     @classmethod
     def _strip_protocol(cls, path):
         path = stringify_path(path)
-        if path.startswith("file://"):
-            path = path[7:]
-        path = os.path.expanduser(path)
-        return make_path_posix(path)
+        _, _, path = path.partition("://")
+        return path
 
     def ls(self, path, detail=False, **kwargs):
-        path = self._strip_protocol(path)
-        paths = [posixpath.join(path, f) for f in os.listdir(path)]
+        from pyarrow.fs import FileSelector
+
+        entries = [
+            self._make_entry(entry) for entry in self.get_file_info(FileSelector(path))
+        ]
         if detail:
-            return [self.info(f) for f in paths]
+            return entries
         else:
-            return paths
+            return [entry["name"] for entry in entries]
 
     def info(self, path, **kwargs):
         path = self._strip_protocol(path)

From 2450e572a6d1a3d3f8f4c2184974beee72c70daa Mon Sep 17 00:00:00 2001
From: Batuhan Taskaya <batuhan@python.org>
Date: Mon, 13 Sep 2021 16:34:24 +0300
Subject: [PATCH 12/21] arrow: fallback on base when _parent() is called

---
 fsspec/implementations/arrow.py | 12 +-----------
 1 file changed, 1 insertion(+), 11 deletions(-)

diff --git a/fsspec/implementations/arrow.py b/fsspec/implementations/arrow.py
index 74d887f94..214c53347 100644
--- a/fsspec/implementations/arrow.py
+++ b/fsspec/implementations/arrow.py
@@ -13,9 +13,7 @@ class ArrowFSWrapper(AbstractFileSystem):
 
     """
 
-    # root_marker = "/"
-    # protocol = "file"
-    # local_file = True
+    root_marker = "/"
 
     def __init__(self, fs, **kwargs):
         super().__init__(**kwargs)
@@ -30,14 +28,6 @@ def __init__(self, fs, **kwargs):
             )
         self.fs = fs
 
-    @classmethod
-    def _parent(cls, path):
-        path = cls._strip_protocol(path).rstrip("/")
-        if "/" in path:
-            return path.rsplit("/", 1)[0]
-        else:
-            return cls.root_marker
-
     @classmethod
     def _strip_protocol(cls, path):
         path = stringify_path(path)

From a384fb2c4248f22571a79981ab38fb4319023896 Mon Sep 17 00:00:00 2001
From: Batuhan Taskaya <batuhan@python.org>
Date: Mon, 13 Sep 2021 18:26:48 +0300
Subject: [PATCH 13/21] arrow: implement ArrowFile, mirror_from, and exception
 management

---
 fsspec/implementations/arrow.py | 112 ++++++++++++++++++++++++++++----
 fsspec/utils.py                 |  19 ++++++
 2 files changed, 119 insertions(+), 12 deletions(-)

diff --git a/fsspec/implementations/arrow.py b/fsspec/implementations/arrow.py
index 214c53347..af4c3fb55 100644
--- a/fsspec/implementations/arrow.py
+++ b/fsspec/implementations/arrow.py
@@ -1,7 +1,32 @@
-from contextlib import closing
+import errno
+import io
+import os
+import secrets
+import shutil
+from contextlib import suppress
+from functools import wraps
 
 from fsspec import AbstractFileSystem
-from fsspec.utils import stringify_path
+from fsspec.utils import mirror_from, stringify_path
+
+
+def wrap_exceptions(func):
+    @wraps(func)
+    def wrapper(*args, **kwargs):
+        try:
+            return func(*args, **kwargs)
+        except OSError as exception:
+            if not exception.args:
+                raise
+
+            message, *args = exception.args
+            if "file does not exist" in message:
+                _, _, path = message.partition(": ")
+                raise FileNotFoundError(errno.ENOENT, os.strerror(errno.ENOENT), path)
+            else:
+                raise
+
+    return wrapper
 
 
 class ArrowFSWrapper(AbstractFileSystem):
@@ -31,14 +56,17 @@ def __init__(self, fs, **kwargs):
     @classmethod
     def _strip_protocol(cls, path):
         path = stringify_path(path)
-        _, _, path = path.partition("://")
+        if "://" in path:
+            _, _, path = path.partition("://")
+
         return path
 
     def ls(self, path, detail=False, **kwargs):
         from pyarrow.fs import FileSelector
 
         entries = [
-            self._make_entry(entry) for entry in self.get_file_info(FileSelector(path))
+            self._make_entry(entry)
+            for entry in self.fs.get_file_info(FileSelector(path))
         ]
         if detail:
             return entries
@@ -57,6 +85,8 @@ def _make_entry(self, info):
             kind = "directory"
         elif info.type is FileType.File:
             kind = "file"
+        elif info.type is FileType.NotFound:
+            raise FileNotFoundError(errno.ENOENT, os.strerror(errno.ENOENT), info.path)
         else:
             kind = "other"
 
@@ -67,26 +97,34 @@ def _make_entry(self, info):
             "mtime": info.mtime,
         }
 
+    @wrap_exceptions
     def cp_file(self, path1, path2, **kwargs):
         path1 = self._strip_protocol(path1).rstrip("/")
         path2 = self._strip_protocol(path2).rstrip("/")
-        self.fs.copy_file(path1, path2)
-
-    def get_file(self, path1, path2, **kwargs):
-        return self.cp_file(path1, path2, **kwargs)
-
-    def put_file(self, path1, path2, **kwargs):
-        return self.cp_file(path1, path2, **kwargs)
 
+        with self._open(path1, "rb") as lstream:
+            tmp_fname = "/".join([self._parent(path2), f".tmp.{secrets.token_hex(16)}"])
+            try:
+                with self.open(tmp_fname, "wb") as rstream:
+                    shutil.copyfileobj(lstream, rstream)
+                self.fs.move(tmp_fname, path2)
+            except BaseException:  # noqa
+                with suppress(FileNotFoundError):
+                    self.fs.delete_file(tmp_fname)
+                raise
+
+    @wrap_exceptions
     def mv_file(self, path1, path2, **kwargs):
         path1 = self._strip_protocol(path1).rstrip("/")
         path2 = self._strip_protocol(path2).rstrip("/")
         self.fs.move(path1, path2)
 
+    @wrap_exceptions
     def rm_file(self, path):
         path = self._strip_protocol(path)
         self.fs.delete_file(path)
 
+    @wrap_exceptions
     def rm(self, path, recursive=False, maxdepth=None):
         path = self._strip_protocol(path).rstrip("/")
         if recursive and self.isdir(path):
@@ -94,6 +132,7 @@ def rm(self, path, recursive=False, maxdepth=None):
         else:
             self.fs.delete_file(path)
 
+    @wrap_exceptions
     def _open(self, path, mode="rb", block_size=None, **kwargs):
         if mode == "rb":
             stream = self.fs.open_input_stream(path)
@@ -102,8 +141,9 @@ def _open(self, path, mode="rb", block_size=None, **kwargs):
         else:
             raise ValueError(f"unsupported mode for Arrow filesystem: {mode!r}")
 
-        return closing(stream)
+        return ArrowFile(self, stream, path, mode, block_size, **kwargs)
 
+    @wrap_exceptions
     def mkdir(self, path, create_parents=True, **kwargs):
         path = self._strip_protocol(path)
         if create_parents:
@@ -111,10 +151,58 @@ def mkdir(self, path, create_parents=True, **kwargs):
         else:
             self.fs.create_dir(path, recursive=False)
 
+    @wrap_exceptions
     def makedirs(self, path, exist_ok=False):
         path = self._strip_protocol(path)
         self.fs.create_dir(path, recursive=True)
 
+    @wrap_exceptions
     def rmdir(self, path):
         path = self._strip_protocol(path)
         self.fs.delete_dir(path)
+
+
+@mirror_from(
+    "stream", ["read", "seek", "tell", "write", "readable", "writable", "close"]
+)
+class ArrowFile(io.IOBase):
+    def __init__(self, fs, stream, path, mode, block_size=None, **kwargs):
+        self.path = path
+        self.mode = mode
+
+        self.fs = fs
+        self.stream = stream
+
+        self.blocksize = self.block_size = block_size
+        self.kwargs = kwargs
+
+    def __enter__(self):
+        return self
+
+    def __exit__(self, *args):
+        return self.close()
+
+
+class HadoopFileSystemWrapper(ArrowFSWrapper):
+
+    protocol = "hdfs"
+
+    def __init__(
+        self,
+        host="default",
+        port=0,
+        user=None,
+        kerb_ticket=None,
+        extra_conf=None,
+        **kwargs,
+    ):
+        from pyarrow.fs import HadoopFileSystem
+
+        fs = HadoopFileSystem(
+            host=host,
+            port=port,
+            user=user,
+            kerb_ticket=kerb_ticket,
+            extra_conf=extra_conf,
+        )
+        super().__init__(fs=fs, **kwargs)
diff --git a/fsspec/utils.py b/fsspec/utils.py
index c2bee8315..235c8799b 100644
--- a/fsspec/utils.py
+++ b/fsspec/utils.py
@@ -4,6 +4,7 @@
 import pathlib
 import re
 import sys
+from functools import partial
 from hashlib import md5
 from urllib.parse import urlsplit
 
@@ -447,3 +448,21 @@ def setup_logging(logger=None, logger_name=None, level="DEBUG", clear=True):
     logger.addHandler(handle)
     logger.setLevel(level)
     return logger
+
+
+def mirror_from(origin_name, methods):
+    """Mirror attributes and methods from the given
+    origin_name attribute of the instance to the
+    decorated class"""
+
+    def origin_getter(method, self):
+        origin = getattr(self, origin_name)
+        return getattr(origin, method)
+
+    def wrapper(cls):
+        for method in methods:
+            wrapped_method = partial(origin_getter, method)
+            setattr(cls, method, property(wrapped_method))
+        return cls
+
+    return wrapper

From a3afee578cf57e4767eee3aa7afa8001c0fd9b22 Mon Sep 17 00:00:00 2001
From: Batuhan Taskaya <batuhan@python.org>
Date: Mon, 13 Sep 2021 20:19:46 +0300
Subject: [PATCH 14/21] arrow: implement clean exists(), and other minor stuff

---
 fsspec/implementations/arrow.py | 24 ++++++++++++------------
 1 file changed, 12 insertions(+), 12 deletions(-)

diff --git a/fsspec/implementations/arrow.py b/fsspec/implementations/arrow.py
index af4c3fb55..e54810447 100644
--- a/fsspec/implementations/arrow.py
+++ b/fsspec/implementations/arrow.py
@@ -6,7 +6,7 @@
 from contextlib import suppress
 from functools import wraps
 
-from fsspec import AbstractFileSystem
+from fsspec.spec import AbstractFileSystem
 from fsspec.utils import mirror_from, stringify_path
 
 
@@ -20,7 +20,7 @@ def wrapper(*args, **kwargs):
                 raise
 
             message, *args = exception.args
-            if "file does not exist" in message:
+            if isinstance(message, str) and "file does not exist" in message:
                 _, _, path = message.partition(": ")
                 raise FileNotFoundError(errno.ENOENT, os.strerror(errno.ENOENT), path)
             else:
@@ -41,17 +41,8 @@ class ArrowFSWrapper(AbstractFileSystem):
     root_marker = "/"
 
     def __init__(self, fs, **kwargs):
-        super().__init__(**kwargs)
-        try:
-            import pyarrow.fs
-        except ImportError:
-            raise ImportError("pyarrow required to use the ArrowFSWrapper")
-
-        if not isinstance(fs, pyarrow.fs.FileSystem):
-            raise TypeError(
-                "'fs' should be an instance of a pyarrow.fs.FileSystem subclass"
-            )
         self.fs = fs
+        super().__init__(**kwargs)
 
     @classmethod
     def _strip_protocol(cls, path):
@@ -78,6 +69,15 @@ def info(self, path, **kwargs):
         [info] = self.fs.get_file_info([path])
         return self._make_entry(info)
 
+    def exists(self, path):
+        path = self._strip_protocol(path)
+        try:
+            self.info(path)
+        except FileNotFoundError:
+            return False
+        else:
+            return True
+
     def _make_entry(self, info):
         from pyarrow.fs import FileType
 

From d6758c1f321bb9db869cd9743532e216a3ab26c6 Mon Sep 17 00:00:00 2001
From: Batuhan Taskaya <batuhan@python.org>
Date: Mon, 13 Sep 2021 20:45:01 +0300
Subject: [PATCH 15/21] arrow: implement get_kwargs_from_url

---
 fsspec/implementations/arrow.py | 14 +++++++++++++-
 1 file changed, 13 insertions(+), 1 deletion(-)

diff --git a/fsspec/implementations/arrow.py b/fsspec/implementations/arrow.py
index e54810447..cbd16763b 100644
--- a/fsspec/implementations/arrow.py
+++ b/fsspec/implementations/arrow.py
@@ -7,7 +7,7 @@
 from functools import wraps
 
 from fsspec.spec import AbstractFileSystem
-from fsspec.utils import mirror_from, stringify_path
+from fsspec.utils import infer_storage_options, mirror_from, stringify_path
 
 
 def wrap_exceptions(func):
@@ -206,3 +206,15 @@ def __init__(
             extra_conf=extra_conf,
         )
         super().__init__(fs=fs, **kwargs)
+
+    @staticmethod
+    def _get_kwargs_from_urls(path):
+        ops = infer_storage_options(path)
+        out = {}
+        if ops.get("host", None):
+            out["host"] = ops["host"]
+        if ops.get("username", None):
+            out["user"] = ops["username"]
+        if ops.get("port", None):
+            out["port"] = ops["port"]
+        return out

From d0599efa6cb1ded5eb1c675501f3feb8c4e51fc6 Mon Sep 17 00:00:00 2001
From: Batuhan Taskaya <batuhan@python.org>
Date: Mon, 13 Sep 2021 21:02:20 +0300
Subject: [PATCH 16/21] arrow: add basic fs tests

---
 fsspec/implementations/arrow.py            |  16 ++-
 fsspec/implementations/tests/test_arrow.py | 160 +++++++++++++++++++++
 2 files changed, 170 insertions(+), 6 deletions(-)
 create mode 100644 fsspec/implementations/tests/test_arrow.py

diff --git a/fsspec/implementations/arrow.py b/fsspec/implementations/arrow.py
index cbd16763b..9d0c410b9 100644
--- a/fsspec/implementations/arrow.py
+++ b/fsspec/implementations/arrow.py
@@ -20,9 +20,8 @@ def wrapper(*args, **kwargs):
                 raise
 
             message, *args = exception.args
-            if isinstance(message, str) and "file does not exist" in message:
-                _, _, path = message.partition(": ")
-                raise FileNotFoundError(errno.ENOENT, os.strerror(errno.ENOENT), path)
+            if isinstance(message, str) and "does not exist" in message:
+                raise FileNotFoundError(errno.ENOENT, message) from exception
             else:
                 raise
 
@@ -114,11 +113,13 @@ def cp_file(self, path1, path2, **kwargs):
                 raise
 
     @wrap_exceptions
-    def mv_file(self, path1, path2, **kwargs):
+    def mv(self, path1, path2, **kwargs):
         path1 = self._strip_protocol(path1).rstrip("/")
         path2 = self._strip_protocol(path2).rstrip("/")
         self.fs.move(path1, path2)
 
+    mv_file = mv
+
     @wrap_exceptions
     def rm_file(self, path):
         path = self._strip_protocol(path)
@@ -127,8 +128,11 @@ def rm_file(self, path):
     @wrap_exceptions
     def rm(self, path, recursive=False, maxdepth=None):
         path = self._strip_protocol(path).rstrip("/")
-        if recursive and self.isdir(path):
-            self.fs.delete_dir(path)
+        if self.isdir(path):
+            if recursive:
+                self.fs.delete_dir(path)
+            else:
+                raise ValueError("Can't delete directories without recursive=False")
         else:
             self.fs.delete_file(path)
 
diff --git a/fsspec/implementations/tests/test_arrow.py b/fsspec/implementations/tests/test_arrow.py
new file mode 100644
index 000000000..e0c535c30
--- /dev/null
+++ b/fsspec/implementations/tests/test_arrow.py
@@ -0,0 +1,160 @@
+import secrets
+
+import pytest
+from pyarrow.fs import FileSystem
+
+from fsspec.implementations.arrow import ArrowFSWrapper
+
+
+@pytest.fixture(scope="function")
+def fs():
+    fs, _ = FileSystem.from_uri("mock://")
+    return ArrowFSWrapper(fs)
+
+
+@pytest.fixture(scope="function")
+def remote_dir(fs):
+    directory = secrets.token_hex(16)
+    fs.makedirs(directory)
+    yield directory
+    fs.rm(directory, recursive=True)
+
+
+def strip_keys(original_entry):
+    entry = original_entry.copy()
+    entry.pop("mtime")
+    return entry
+
+
+def test_info(fs, remote_dir):
+    fs.touch(remote_dir + "/a.txt")
+    details = fs.info(remote_dir + "/a.txt")
+    assert details["type"] == "file"
+    assert details["name"] == remote_dir + "/a.txt"
+    assert details["size"] == 0
+
+    fs.mkdir(remote_dir + "/dir")
+    details = fs.info(remote_dir + "/dir")
+    assert details["type"] == "directory"
+    assert details["name"] == remote_dir + "/dir"
+
+    details = fs.info(remote_dir + "/dir/")
+    assert details["name"] == remote_dir + "/dir/"
+
+
+def test_move(fs, remote_dir):
+    fs.touch(remote_dir + "/a.txt")
+    initial_info = fs.info(remote_dir + "/a.txt")
+
+    fs.move(remote_dir + "/a.txt", remote_dir + "/b.txt")
+    secondary_info = fs.info(remote_dir + "/b.txt")
+
+    assert not fs.exists(remote_dir + "/a.txt")
+    assert fs.exists(remote_dir + "/b.txt")
+
+    initial_info.pop("name")
+    secondary_info.pop("name")
+    assert initial_info == secondary_info
+
+
+def test_copy(fs, remote_dir):
+    fs.touch(remote_dir + "/a.txt")
+    initial_info = fs.info(remote_dir + "/a.txt")
+
+    fs.copy(remote_dir + "/a.txt", remote_dir + "/b.txt")
+    secondary_info = fs.info(remote_dir + "/b.txt")
+
+    assert fs.exists(remote_dir + "/a.txt")
+    assert fs.exists(remote_dir + "/b.txt")
+
+    initial_info.pop("name")
+    secondary_info.pop("name")
+    assert strip_keys(initial_info) == strip_keys(secondary_info)
+
+
+def test_rm(fs, remote_dir):
+    fs.touch(remote_dir + "/a.txt")
+    fs.rm(remote_dir + "/a.txt", recursive=True)
+    assert not fs.exists(remote_dir + "/a.txt")
+
+    fs.mkdir(remote_dir + "/dir")
+    fs.rm(remote_dir + "/dir", recursive=True)
+    assert not fs.exists(remote_dir + "/dir")
+
+    fs.mkdir(remote_dir + "/dir")
+    fs.touch(remote_dir + "/dir/a")
+    fs.touch(remote_dir + "/dir/b")
+    fs.mkdir(remote_dir + "/dir/c/")
+    fs.touch(remote_dir + "/dir/c/a/")
+    fs.rm(remote_dir + "/dir", recursive=True)
+    assert not fs.exists(remote_dir + "/dir")
+
+
+def test_ls(fs, remote_dir):
+    fs.mkdir(remote_dir + "dir/")
+    files = set()
+    for no in range(8):
+        file = remote_dir + f"dir/test_{no}"
+        fs.touch(file)
+        files.add(file)
+
+    assert set(fs.ls(remote_dir + "dir/")) == files
+
+    dirs = fs.ls(remote_dir + "dir/", detail=True)
+    expected = [fs.info(file) for file in files]
+
+    by_name = lambda details: details["name"]
+    dirs.sort(key=by_name)
+    expected.sort(key=by_name)
+
+    assert dirs == expected
+
+
+def test_mkdir(fs, remote_dir):
+    fs.mkdir(remote_dir + "dir/")
+    assert fs.isdir(remote_dir + "dir/")
+    assert len(fs.ls(remote_dir + "dir/")) == 0
+
+    fs.mkdir(remote_dir + "dir/sub", create_parents=False)
+    assert fs.isdir(remote_dir + "dir/sub")
+
+
+def test_makedirs(fs, remote_dir):
+    fs.makedirs(remote_dir + "dir/a/b/c/")
+    assert fs.isdir(remote_dir + "dir/a/b/c/")
+    assert fs.isdir(remote_dir + "dir/a/b/")
+    assert fs.isdir(remote_dir + "dir/a/")
+
+    fs.makedirs(remote_dir + "dir/a/b/c/", exist_ok=True)
+
+
+def test_exceptions(fs, remote_dir):
+    with pytest.raises(FileNotFoundError):
+        with fs.open(remote_dir + "/a.txt"):
+            ...
+
+    with pytest.raises(FileNotFoundError):
+        fs.copy(remote_dir + "/u.txt", remote_dir + "/y.txt")
+
+
+def test_open_rw(fs, remote_dir):
+    data = b"dvc.org"
+
+    with fs.open(remote_dir + "/a.txt", "wb") as stream:
+        stream.write(data)
+
+    with fs.open(remote_dir + "/a.txt") as stream:
+        assert stream.read() == data
+
+
+def test_open_rw_flush(fs, remote_dir):
+    data = b"dvc.org"
+
+    with fs.open(remote_dir + "/b.txt", "wb") as stream:
+        for _ in range(200):
+            stream.write(data)
+            stream.write(data)
+            stream.flush()
+
+    with fs.open(remote_dir + "/b.txt", "rb") as stream:
+        assert stream.read() == data * 400

From 0f9c36a4aeabe708151d95474005fbfdb6fa6926 Mon Sep 17 00:00:00 2001
From: Batuhan Taskaya <batuhan@python.org>
Date: Mon, 13 Sep 2021 21:06:27 +0300
Subject: [PATCH 17/21] arrow: register / document

---
 docs/source/api.rst             |  2 ++
 fsspec/implementations/arrow.py | 17 +++++++++++++++++
 fsspec/registry.py              |  4 ++++
 setup.py                        |  1 +
 4 files changed, 24 insertions(+)

diff --git a/docs/source/api.rst b/docs/source/api.rst
index 456839119..d2131b82d 100644
--- a/docs/source/api.rst
+++ b/docs/source/api.rst
@@ -97,6 +97,8 @@ Built-in Implementations
 .. autosummary::
    fsspec.implementations.ftp.FTPFileSystem
    fsspec.implementations.hdfs.PyArrowHDFS
+   fsspec.implementations.arrow.ArrowFSWrapper
+   fsspec.implementations.arrow.HadoopFileSystemWrapper
    fsspec.implementations.dask.DaskWorkerFileSystem
    fsspec.implementations.http.HTTPFileSystem
    fsspec.implementations.local.LocalFileSystem
diff --git a/fsspec/implementations/arrow.py b/fsspec/implementations/arrow.py
index 9d0c410b9..9635ef94f 100644
--- a/fsspec/implementations/arrow.py
+++ b/fsspec/implementations/arrow.py
@@ -188,6 +188,8 @@ def __exit__(self, *args):
 
 
 class HadoopFileSystemWrapper(ArrowFSWrapper):
+    """A wrapper on top of the pyarrow.fs.HadoopFileSystem
+    to connect it's interface with fsspec"""
 
     protocol = "hdfs"
 
@@ -200,6 +202,21 @@ def __init__(
         extra_conf=None,
         **kwargs,
     ):
+        """
+
+        Parameters
+        ----------
+        host: str
+            Hostname, IP or "default" to try to read from Hadoop config
+        port: int
+            Port to connect on, or default from Hadoop config if 0
+        user: str or None
+            If given, connect as this username
+        kerb_ticket: str or None
+            If given, use this ticket for authentication
+        extra_conf: None or dict
+            Passed on to HadoopFileSystem
+        """
         from pyarrow.fs import HadoopFileSystem
 
         fs = HadoopFileSystem(
diff --git a/fsspec/registry.py b/fsspec/registry.py
index 18cfda2d8..30ba93217 100644
--- a/fsspec/registry.py
+++ b/fsspec/registry.py
@@ -129,6 +129,10 @@ def register_implementation(name, cls, clobber=True, errtxt=None):
         "class": "fsspec.implementations.hdfs.PyArrowHDFS",
         "err": "pyarrow and local java libraries required for HDFS",
     },
+    "arrow_hdfs": {
+        "class": "fsspec.implementations.arrow.HadoopFileSystemWrapper",
+        "err": "pyarrow and local java libraries required for HDFS",
+    },
     "webhdfs": {
         "class": "fsspec.implementations.webhdfs.WebHDFS",
         "err": 'webHDFS access requires "requests" to be installed',
diff --git a/setup.py b/setup.py
index 18b7b48f2..23bbae30f 100644
--- a/setup.py
+++ b/setup.py
@@ -45,6 +45,7 @@
         "github": ["requests"],
         "gs": ["gcsfs"],
         "hdfs": ["pyarrow >= 1"],
+        "arrow": ["pyarrow >= 1"],
         "http": ["requests", "aiohttp"],
         "sftp": ["paramiko"],
         "s3": ["s3fs"],

From 29196794ef50ea2a4e32dd956bf8c7b717d9a320 Mon Sep 17 00:00:00 2001
From: Batuhan Taskaya <batuhan@python.org>
Date: Mon, 13 Sep 2021 21:18:02 +0300
Subject: [PATCH 18/21] arrow: skip when pyarrow is not available

---
 fsspec/implementations/tests/test_arrow.py | 6 ++++--
 1 file changed, 4 insertions(+), 2 deletions(-)

diff --git a/fsspec/implementations/tests/test_arrow.py b/fsspec/implementations/tests/test_arrow.py
index e0c535c30..b623131e6 100644
--- a/fsspec/implementations/tests/test_arrow.py
+++ b/fsspec/implementations/tests/test_arrow.py
@@ -1,9 +1,11 @@
 import secrets
 
 import pytest
-from pyarrow.fs import FileSystem
 
-from fsspec.implementations.arrow import ArrowFSWrapper
+pyarrow_fs = pytest.importorskip("pyarrow.fs")
+FileSystem = pyarrow_fs.FileSystem
+
+from fsspec.implementations.arrow import ArrowFSWrapper  # noqa
 
 
 @pytest.fixture(scope="function")

From f0544aaffc101f69fe1bb38b6cce18964b457725 Mon Sep 17 00:00:00 2001
From: Batuhan Taskaya <batuhan@python.org>
Date: Tue, 14 Sep 2021 17:47:04 +0300
Subject: [PATCH 19/21] arrow: drop Wrapper from HadoopFileSystem

---
 docs/source/api.rst             | 2 +-
 fsspec/implementations/arrow.py | 2 +-
 fsspec/registry.py              | 2 +-
 3 files changed, 3 insertions(+), 3 deletions(-)

diff --git a/docs/source/api.rst b/docs/source/api.rst
index d2131b82d..2692983c6 100644
--- a/docs/source/api.rst
+++ b/docs/source/api.rst
@@ -98,7 +98,7 @@ Built-in Implementations
    fsspec.implementations.ftp.FTPFileSystem
    fsspec.implementations.hdfs.PyArrowHDFS
    fsspec.implementations.arrow.ArrowFSWrapper
-   fsspec.implementations.arrow.HadoopFileSystemWrapper
+   fsspec.implementations.arrow.HadoopFileSystem
    fsspec.implementations.dask.DaskWorkerFileSystem
    fsspec.implementations.http.HTTPFileSystem
    fsspec.implementations.local.LocalFileSystem
diff --git a/fsspec/implementations/arrow.py b/fsspec/implementations/arrow.py
index 9635ef94f..0a785bda1 100644
--- a/fsspec/implementations/arrow.py
+++ b/fsspec/implementations/arrow.py
@@ -187,7 +187,7 @@ def __exit__(self, *args):
         return self.close()
 
 
-class HadoopFileSystemWrapper(ArrowFSWrapper):
+class HadoopFileSystem(ArrowFSWrapper):
     """A wrapper on top of the pyarrow.fs.HadoopFileSystem
     to connect it's interface with fsspec"""
 
diff --git a/fsspec/registry.py b/fsspec/registry.py
index 30ba93217..37f880343 100644
--- a/fsspec/registry.py
+++ b/fsspec/registry.py
@@ -130,7 +130,7 @@ def register_implementation(name, cls, clobber=True, errtxt=None):
         "err": "pyarrow and local java libraries required for HDFS",
     },
     "arrow_hdfs": {
-        "class": "fsspec.implementations.arrow.HadoopFileSystemWrapper",
+        "class": "fsspec.implementations.arrow.HadoopFileSystem",
         "err": "pyarrow and local java libraries required for HDFS",
     },
     "webhdfs": {

From a02969a919dd73e49c16e997d2d6e453b1f48a51 Mon Sep 17 00:00:00 2001
From: Batuhan Taskaya <batuhan@python.org>
Date: Tue, 14 Sep 2021 17:50:29 +0300
Subject: [PATCH 20/21] arrow: add test_move_recursive

---
 fsspec/implementations/tests/test_arrow.py | 21 +++++++++++++++++++++
 1 file changed, 21 insertions(+)

diff --git a/fsspec/implementations/tests/test_arrow.py b/fsspec/implementations/tests/test_arrow.py
index b623131e6..5fb519b16 100644
--- a/fsspec/implementations/tests/test_arrow.py
+++ b/fsspec/implementations/tests/test_arrow.py
@@ -59,6 +59,27 @@ def test_move(fs, remote_dir):
     assert initial_info == secondary_info
 
 
+def test_move_recursive(fs, remote_dir):
+    src = remote_dir + "/src"
+    dest = remote_dir + "/dest"
+
+    assert fs.isdir(src) is False
+    fs.mkdir(src)
+    assert fs.isdir(src)
+
+    fs.touch(src + "/a.txt")
+    fs.mkdir(src + "/b")
+    fs.touch(src + "/b/c.txt")
+    fs.move(src, dest, recursive=True)
+
+    assert fs.isdir(src) is False
+    assert not fs.exists(src)
+
+    assert fs.isdir(dest)
+    assert fs.exists(dest)
+    assert fs.cat(dest + "/b/c.txt") == fs.cat(dest + "/a.txt") == b""
+
+
 def test_copy(fs, remote_dir):
     fs.touch(remote_dir + "/a.txt")
     initial_info = fs.info(remote_dir + "/a.txt")

From b0c16d9a82783f04c643207ac268842cea45815a Mon Sep 17 00:00:00 2001
From: Batuhan Taskaya <batuhan@python.org>
Date: Tue, 14 Sep 2021 18:42:26 +0300
Subject: [PATCH 21/21] arrow: test for mirror_from

---
 docs/source/api.rst        |  6 ++++++
 fsspec/tests/test_utils.py | 32 ++++++++++++++++++++++++++++++++
 2 files changed, 38 insertions(+)

diff --git a/docs/source/api.rst b/docs/source/api.rst
index 2692983c6..48e03a644 100644
--- a/docs/source/api.rst
+++ b/docs/source/api.rst
@@ -123,6 +123,12 @@ Built-in Implementations
 .. autoclass:: fsspec.implementations.hdfs.PyArrowHDFS
    :members: __init__
 
+.. autoclass:: fsspec.implementations.hdfs.ArrowFSWrapper
+   :members: __init__
+
+.. autoclass:: fsspec.implementations.hdfs.HadoopFileSystem
+   :members: __init__
+
 .. autoclass:: fsspec.implementations.dask.DaskWorkerFileSystem
    :members: __init__
 
diff --git a/fsspec/tests/test_utils.py b/fsspec/tests/test_utils.py
index 1d4a89a94..593568cc5 100644
--- a/fsspec/tests/test_utils.py
+++ b/fsspec/tests/test_utils.py
@@ -1,5 +1,6 @@
 import io
 import sys
+from unittest.mock import Mock
 
 import pytest
 
@@ -7,6 +8,7 @@
     can_be_local,
     common_prefix,
     infer_storage_options,
+    mirror_from,
     other_paths,
     read_block,
     seek_delimiter,
@@ -310,3 +312,33 @@ def test_log():
 def test_can_local(par):
     url, outcome = par
     assert can_be_local(url) == outcome
+
+
+def test_mirror_from():
+
+    mock = Mock()
+    mock.attr = 1
+
+    @mirror_from("client", ["attr", "func_1", "func_2"])
+    class Real:
+        @property
+        def client(self):
+            return mock
+
+        def func_2(self):
+            assert False, "have to overwrite this"
+
+        def func_3(self):
+            return "should succeed"
+
+    obj = Real()
+    assert obj.attr == mock.attr
+
+    obj.func_1()
+    mock.func_1.assert_called()
+
+    obj.func_2(1, 2)
+    mock.func_2.assert_called_with(1, 2)
+
+    assert obj.func_3() == "should succeed"
+    mock.func_3.assert_not_called()