Repository: fsspec/filesystem_spec

Issue:
The arrow_hdfs protocol violates rfc3986 in urllib
According to [rfc3986](https://www.rfc-editor.org/rfc/rfc3986#section-3.1), schema names consist of a sequence of characters starting with a letter and followed by any combination of letters, digits, plus ("+"), period ("."), or hyphen ("-") .

The new HDFS implementation violates this rule as the protocol was defined as `arrow_hdfs` in registry.py: 

https://github.com/fsspec/filesystem_spec/blob/742a103849adf2d0c9f3ad3fe08010d8de688268/fsspec/registry.py#L132

Thus, it is difficult to validate that a url is being provided correctly. For example:

```python
import urllib.parse

urllib.parse.urlsplit("arrow_hdfs://server/path")
SplitResult(scheme='', netloc='', path='arrow_hdfs://server/path', query='', fragment='')
```

This problem can be worked around by changing the protocol name from `arrow_hdfs` to `arrow-hdfs`:

```python
import urllib.parse
urllib.parse.urlsplit("arrow-hdfs://server/path")
SplitResult(scheme='arrow-hdfs', netloc='server', path='/path', query='', fragment='')
```

Discussion:
hallenmaia: @martindurant I think it's ready!
martindurant: Looks very reasonable to me. I'll leave it here a couple of days for anyone else interested.

Patch:
diff --git a/docs/source/api.rst b/docs/source/api.rst
index 1bd3e25..7ec1f6f 100644
--- a/docs/source/api.rst
+++ b/docs/source/api.rst
@@ -110,7 +110,6 @@ Built-in Implementations
 
 .. autosummary::
    fsspec.implementations.ftp.FTPFileSystem
-   fsspec.implementations.hdfs.PyArrowHDFS
    fsspec.implementations.arrow.ArrowFSWrapper
    fsspec.implementations.arrow.HadoopFileSystem
    fsspec.implementations.dask.DaskWorkerFileSystem
@@ -135,9 +134,6 @@ Built-in Implementations
 .. autoclass:: fsspec.implementations.ftp.FTPFileSystem
    :members: __init__
 
-.. autoclass:: fsspec.implementations.hdfs.PyArrowHDFS
-   :members: __init__
-
 .. autoclass:: fsspec.implementations.arrow.ArrowFSWrapper
    :members: __init__
 
diff --git a/fsspec/implementations/hdfs.py b/fsspec/implementations/hdfs.py
deleted file mode 100644
index 2a2e23d..0000000
--- a/fsspec/implementations/hdfs.py
+++ /dev/null
@@ -1,267 +0,0 @@
-import secrets
-import shutil
-import weakref
-from contextlib import suppress
-
-from pyarrow.hdfs import HadoopFileSystem
-
-from fsspec.implementations.arrow import wrap_exceptions
-from fsspec.spec import AbstractFileSystem
-from fsspec.utils import infer_storage_options, mirror_from
-
-
-@mirror_from(
-    "client",
-    [
-        "chmod",
-        "chown",
-        "user",
-        "df",
-        "disk_usage",
-        "download",
-        "driver",
-        "exists",
-        "extra_conf",
-        "get_capacity",
-        "get_space_used",
-        "host",
-        "is_open",
-        "kerb_ticket",
-        "strip_protocol",
-        "mkdir",
-        "port",
-        "get_capacity",
-        "get_space_used",
-        "df",
-        "chmod",
-        "chown",
-        "disk_usage",
-        "download",
-        "read_parquet",
-        "rm",
-        "stat",
-        "upload",
-    ],
-)
-class PyArrowHDFS(AbstractFileSystem):
-    """Adapted version of Arrow's HadoopFileSystem
-
-    This is a very simple wrapper over the pyarrow.hdfs.HadoopFileSystem, which
-    passes on all calls to the underlying class."""
-
-    protocol = "hdfs", "file"
-
-    def __init__(
-        self,
-        host="default",
-        port=0,
-        user=None,
-        kerb_ticket=None,
-        driver="libhdfs",
-        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
-        driver: 'libhdfs' or 'libhdfs3'
-            Binary driver; libhdfs if the JNI library and default
-        extra_conf: None or dict
-            Passed on to HadoopFileSystem
-        """
-        super().__init__(**kwargs)
-
-        self.client = HadoopFileSystem(
-            host=host,
-            port=port,
-            user=user,
-            kerb_ticket=kerb_ticket,
-            driver=driver,
-            extra_conf=extra_conf,
-        )
-        weakref.finalize(self, lambda: self.client.close())
-
-        self.pars = (host, port, user, kerb_ticket, driver, extra_conf)
-
-    @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
-
-    @classmethod
-    def _strip_protocol(cls, path):
-        ops = infer_storage_options(path)
-        path = ops["path"]
-        # infer_store_options leaves file:/ prefixes alone
-        # for local hdfs instances
-        if path.startswith("file:"):
-            path = path[5:]
-        return path
-
-    def __reduce_ex__(self, protocol):
-        return PyArrowHDFS, self.pars
-
-    def close(self):
-        self.client.close()
-
-    @wrap_exceptions
-    def ls(self, path, detail=True):
-        listing = [
-            self._adjust_entry(entry) for entry in self.client.ls(path, detail=True)
-        ]
-
-        if detail:
-            return listing
-        else:
-            return [entry["name"] for entry in listing]
-
-    @wrap_exceptions
-    def info(self, path):
-        return self._adjust_entry(self.client.info(path))
-
-    def _adjust_entry(self, original_entry):
-        entry = original_entry.copy()
-        if "type" not in entry:
-            if "kind" in entry:
-                entry["type"] = entry["kind"]
-        if "name" not in entry:
-            if "path" in entry:
-                entry["name"] = entry["path"]
-
-        if "name" in entry:
-            entry["name"] = self._strip_protocol(entry["name"])
-        return entry
-
-    @wrap_exceptions
-    def cp_file(self, lpath, rpath, **kwargs):
-        if self.isdir(lpath):
-            self.makedirs(rpath)
-            return
-
-        with self.open(lpath) as lstream:
-            tmp_fname = "/".join([self._parent(rpath), f".tmp.{secrets.token_hex(16)}"])
-            # Perform an atomic copy (stream to a temporary file and
-            # move it to the actual destination).
-            try:
-                with self.open(tmp_fname, "wb") as rstream:
-                    shutil.copyfileobj(lstream, rstream)
-                self.client.mv(tmp_fname, rpath)
-            except BaseException:  # noqa
-                with suppress(FileNotFoundError):
-                    self.client.rm(tmp_fname)
-                raise
-
-    @wrap_exceptions
-    def rm_file(self, path):
-        return self.client.rm(path)
-
-    @wrap_exceptions
-    def makedirs(self, path, exist_ok=False):
-        if not exist_ok and self.exists(path):
-            raise FileExistsError(path)
-
-        return self.client.mkdir(path, create_parents=True)
-
-    @wrap_exceptions
-    def _open(
-        self,
-        path,
-        mode="rb",
-        block_size=None,
-        autocommit=True,
-        cache_options=None,
-        **kwargs,
-    ):
-        """
-
-        Parameters
-        ----------
-        path: str
-            Location of file; should start with '/'
-        mode: str
-        block_size: int
-            Hadoop block size, e.g., 2**26
-        autocommit: True
-            Transactions are not yet implemented for HDFS; errors if not True
-        kwargs: dict or None
-            Hadoop config parameters
-
-        Returns
-        -------
-        HDFSFile file-like instance
-        """
-
-        return HDFSFile(
-            self,
-            path,
-            mode,
-            block_size=block_size,
-            autocommit=autocommit,
-            cache_options=cache_options,
-            **kwargs,
-        )
-
-
-class HDFSFile(object):
-    """Wrapper around arrow's HdfsFile
-
-    Allows seek beyond EOF and (eventually) commit/discard
-    """
-
-    def __init__(
-        self,
-        fs,
-        path,
-        mode,
-        block_size,
-        autocommit=True,
-        cache_type="readahead",
-        cache_options=None,
-        **kwargs,
-    ):
-        # TODO: Inherit from AbstractBufferedFile?
-        if not autocommit:
-            raise NotImplementedError(
-                "HDFSFile cannot be opened with 'autocommit=False'."
-            )
-
-        self.fs = fs
-        self.path = path
-        self.mode = mode
-        self.blocksize = self.block_size = block_size
-        self.fh = fs.client.open(path, mode, block_size, **kwargs)
-        if self.fh.readable():
-            self.seek_size = self.size()
-
-    def seek(self, loc, whence=0):
-        if whence == 0 and self.readable():
-            loc = min(loc, self.seek_size)
-        return self.fh.seek(loc, whence)
-
-    def __getattr__(self, item):
-        return getattr(self.fh, item)
-
-    def __reduce_ex__(self, protocol):
-        return HDFSFile, (self.fs, self.path, self.mode, self.block_size)
-
-    def __enter__(self):
-        return self
-
-    def __exit__(self, exc_type, exc_val, exc_tb):
-        self.close()
diff --git a/fsspec/registry.py b/fsspec/registry.py
index afb8fe6..7ccc488 100644
--- a/fsspec/registry.py
+++ b/fsspec/registry.py
@@ -1,4 +1,5 @@
 import importlib
+import warnings
 
 __all__ = ["registry", "get_filesystem_class", "default"]
 
@@ -126,7 +127,7 @@ known_implementations = {
     },
     "ftp": {"class": "fsspec.implementations.ftp.FTPFileSystem"},
     "hdfs": {
-        "class": "fsspec.implementations.hdfs.PyArrowHDFS",
+        "class": "fsspec.implementations.arrow.HadoopFileSystem",
         "err": "pyarrow and local java libraries required for HDFS",
     },
     "arrow_hdfs": {
@@ -268,6 +269,13 @@ def filesystem(protocol, **storage_options):
     ``storage_options`` are specific to the protocol being chosen, and are
     passed directly to the class.
     """
+    if protocol == "arrow_hdfs":
+        warnings.warn(
+            "The 'arrow_hdfs' protocol has been deprecated and will be "
+            "removed in the future. Specify it as 'hdfs'.",
+            DeprecationWarning,
+        )
+
     cls = get_filesystem_class(protocol)
     return cls(**storage_options)
 
diff --git a/setup.cfg b/setup.cfg
index fb714ab..e813171 100644
--- a/setup.cfg
+++ b/setup.cfg
@@ -25,7 +25,7 @@ ignore =
 
 [isort]
 known_first_party=fsspec
-known_third_party=aiohttp,dask,distributed,fuse,libarchive,numpy,panel,paramiko,pyarrow,pygit2,pytest,requests,s3fs,setuptools,smbclient
+known_third_party=aiohttp,dask,distributed,fuse,libarchive,numpy,panel,paramiko,pygit2,pytest,requests,s3fs,setuptools,smbclient
 multi_line_output=3
 include_trailing_comma=True
 force_grid_wrap=0


Test patch:
diff --git a/fsspec/implementations/tests/test_hdfs.py b/fsspec/implementations/tests/test_hdfs.py
deleted file mode 100644
index 6f710c0..0000000
--- a/fsspec/implementations/tests/test_hdfs.py
+++ /dev/null
@@ -1,139 +0,0 @@
-from pathlib import Path
-
-import pytest
-
-import fsspec
-
-pyarrow = pytest.importorskip("pyarrow")
-
-basedir = "/tmp/test-fsspec"
-data = b"\n".join([b"some test data"] * 1000)
-
-
-@pytest.fixture
-def hdfs(request):
-    try:
-        hdfs = pyarrow.hdfs.HadoopFileSystem()
-    except IOError:
-        pytest.skip("No HDFS configured")
-
-    if hdfs.exists(basedir):
-        hdfs.rm(basedir, recursive=True)
-
-    hdfs.mkdir(basedir)
-
-    with hdfs.open(basedir + "/file", "wb") as f:
-        f.write(data)
-
-    yield hdfs
-
-    if hdfs.exists(basedir):
-        hdfs.rm(basedir, recursive=True)
-
-
-def test_ls(hdfs):
-    fs = fsspec.filesystem("hdfs")
-    fs.touch(basedir + "/file_2")
-    fs.mkdir(basedir + "/dir_1")
-    fs.touch(basedir + "/dir_1/file_1")
-    fs.mkdir(basedir + "/dir_2")
-
-    out = {(f["name"], f["kind"]) for f in fs.ls(basedir)}
-    assert out == {
-        (basedir + "/file", "file"),
-        (basedir + "/file_2", "file"),
-        (basedir + "/dir_1", "directory"),
-        (basedir + "/dir_2", "directory"),
-    }
-
-
-def test_walk(hdfs):
-    h = fsspec.filesystem("hdfs")
-    out = h.walk(basedir)
-    assert list(out) == list(hdfs.walk(basedir))
-
-
-def test_isdir(hdfs):
-    h = fsspec.filesystem("hdfs")
-    assert h.isdir(basedir)
-    assert not h.isdir(basedir + "/file")
-
-
-def test_exists(hdfs):
-    h = fsspec.filesystem("hdfs")
-    assert not h.exists(basedir + "/notafile")
-
-
-def test_read(hdfs):
-    h = fsspec.filesystem("hdfs")
-    out = basedir + "/file"
-    with h.open(out, "rb") as f:
-        assert f.read() == data
-    with h.open(out, "rb", block_size=0) as f:
-        assert f.read() == data
-    with h.open(out, "rb") as f:
-        assert f.read(100) + f.read() == data
-
-
-def test_copy(hdfs):
-    fs = fsspec.filesystem("hdfs")
-
-    fs.mkdir(basedir + "/test_dir")
-    fs.touch(basedir + "/test_dir/a")
-    fs.touch(basedir + "/test_dir/b")
-    fs.mkdir(basedir + "/test_dir/c")
-    fs.touch(basedir + "/test_dir/c/d")
-
-    fs.copy(basedir + "/test_dir", basedir + "/copy_dir", recursive=True)
-    assert fs.find(basedir + "/copy_dir", detail=False) == [
-        basedir + "/copy_dir" + "/a",
-        basedir + "/copy_dir" + "/b",
-        basedir + "/copy_dir" + "/c/d",
-    ]
-
-
-def test_put_get(hdfs, tmpdir):
-    fs = fsspec.filesystem("hdfs")
-
-    src_dir = Path(tmpdir / "source")
-    dst_dir = Path(tmpdir / "destination")
-
-    src_dir.mkdir()
-    (src_dir / "file_1.txt").write_text("file_1")
-    (src_dir / "file_2.txt").write_text("file_2")
-    (src_dir / "dir_1").mkdir()
-    (src_dir / "dir_1" / "file_3.txt").write_text("file_3")
-    (src_dir / "dir_1" / "file_4.txt").write_text("file_4")
-    (src_dir / "dir_2").mkdir()
-
-    fs.put(str(src_dir), basedir + "/src", recursive=True)
-    fs.get(basedir + "/src", str(dst_dir), recursive=True)
-
-    files = [file.relative_to(dst_dir) for file in dst_dir.glob("**/*")]
-
-    assert set(map(str, files)) == {
-        "file_1.txt",
-        "file_2.txt",
-        "dir_1/file_3.txt",
-        "dir_1/file_4.txt",
-        "dir_1",
-        "dir_2",
-    }
-
-    assert {
-        (dst_dir / file).read_text() for file in files if (dst_dir / file).is_file()
-    } == {"file_1", "file_2", "file_3", "file_4"}
-
-
-def test_put_file_get_file(hdfs, tmpdir):
-    fs = fsspec.filesystem("hdfs")
-
-    src_file = Path(tmpdir / "src_file")
-    dst_file = Path(tmpdir / "dst_file")
-
-    src_file.write_bytes(b"heyhey")
-
-    fs.put_file(src_file, basedir + "/src_file")
-    fs.get_file(basedir + "/src_file", dst_file)
-
-    assert dst_file.read_bytes() == b"heyhey"
diff --git a/fsspec/tests/test_registry.py b/fsspec/tests/test_registry.py
index 7bd9584..bb33128 100644
--- a/fsspec/tests/test_registry.py
+++ b/fsspec/tests/test_registry.py
@@ -6,6 +6,7 @@ import pytest
 from fsspec.registry import (
     ReadOnlyError,
     _registry,
+    filesystem,
     get_filesystem_class,
     known_implementations,
     register_implementation,
@@ -104,3 +105,18 @@ def test_entry_points_registered_on_import(clear_registry, clean_imports):
 
         get_filesystem_class("test")
         assert "test" in registry
+
+
+def test_filesystem_warning_arrow_hdfs_deprecated(clear_registry, clean_imports):
+    mock_ep = create_autospec(EntryPoint, module="fsspec.spec.AbstractFileSystem")
+    mock_ep.name = "arrow_hdfs"  # this can't be set in the constructor...
+    mock_ep.value = "fsspec.spec.AbstractFileSystem"
+    if sys.version_info < (3, 8):
+        import_location = "importlib_metadata.entry_points"
+    else:
+        import_location = "importlib.metadata.entry_points"
+    with patch(import_location, return_value={"fsspec.specs": [mock_ep]}):
+        import fsspec  # noqa
+
+        with pytest.warns(DeprecationWarning):
+            filesystem("arrow_hdfs")