Repository: criteo/cluster-pack (language: python)

Issue:
Replace EnhancedFileSystem with fsspec
The current implementation of `EnhancedFileSystem` is based on the legacy pyarrow filesystem interface that was removed in pyarrow 16.0.0 (https://github.com/apache/arrow/pull/39825).

We can entirely replace `EnhancedFileSystem` with [fsspec](https://filesystem-spec.readthedocs.io/en/latest/index.html). For HDFS fsspec relies on the new pyarrow [filesystem interface](https://arrow.apache.org/docs/python/filesystems.html#hadoop-distributed-file-system-hdfs).

Resolves #87


Discussion / hints:

Fix patch:
diff --git a/cluster_pack/filesystem.py b/cluster_pack/filesystem.py
index a03ab33..2c5b068 100644
--- a/cluster_pack/filesystem.py
+++ b/cluster_pack/filesystem.py
@@ -1,16 +1,32 @@
 import logging
 from typing import Tuple, Any
+from urllib.parse import urlparse
 
 import fsspec
-import fsspec.implementations.arrow
-
-# Support viewfs:// protocol for HDFS
-fsspec.register_implementation("viewfs", fsspec.implementations.arrow.HadoopFileSystem)
 
 _logger = logging.getLogger(__name__)
 
 
 def resolve_filesystem_and_path(uri: str, **kwargs: Any) -> Tuple[fsspec.AbstractFileSystem, str]:
-    fs, fs_path = fsspec.url_to_fs(uri, **kwargs)
+    parsed_uri = urlparse(uri)
+    fs_path = parsed_uri.path
+    if parsed_uri.scheme == 'hdfs' or parsed_uri.scheme == 'viewfs':
+        netloc_split = parsed_uri.netloc.split(':')
+        host = netloc_split[0]
+        if host == '':
+            host = 'default'
+        else:
+            host = parsed_uri.scheme + "://" + host
+        port = 0
+        if len(netloc_split) == 2 and netloc_split[1].isnumeric():
+            port = int(netloc_split[1])
+
+        fs = fsspec.filesystem("hdfs", host=host, port=port, **kwargs)
+    elif parsed_uri.scheme == '':
+        # Input is local path such as /home/user/myfile.parquet
+        fs = fsspec.filesystem('file', **kwargs)
+    else:
+        fs = fsspec.filesystem(parsed_uri.scheme, **kwargs)
+
     _logger.info(f"Resolved base filesystem: {type(fs)}")
     return fs, fs_path


Test patch:
diff --git a/tests/test_filesystem.py b/tests/test_filesystem.py
index 068d46a..ae70806 100644
--- a/tests/test_filesystem.py
+++ b/tests/test_filesystem.py
@@ -1,4 +1,6 @@
 import os
+from unittest import mock
+
 import pytest
 import subprocess
 import tempfile
@@ -141,3 +143,24 @@ def test_put():
 
         assert os.path.exists(remote_file)
         assert os.stat(remote_file).st_mode & 0o777 == 0o755
+
+
+@pytest.mark.parametrize(
+    "uri,expected_protocol,expected_kwargs,expected_path",
+    [
+        ("viewfs:///path/", "hdfs", {"host": "default", "port": 0}, "/path/"),
+        ("viewfs://root/path/", "hdfs", {"host": "viewfs://root", "port": 0}, "/path/"),
+        ("viewfs://localhost:1234/path/", "hdfs", {"host": "viewfs://localhost", "port": 1234},
+         "/path/"),
+        ("hdfs://root/path/", "hdfs", {"host": "hdfs://root", "port": 0}, "/path/"),
+        ("/path/", "file", {}, "/path/"),
+        ("file:///path/", "file", {}, "/path/"),
+    ]
+)
+def test_resolve_filesystem_and_path(uri, expected_protocol, expected_kwargs, expected_path):
+    with mock.patch("fsspec.filesystem") as fsspec_filesystem_mock:
+        fs, path = filesystem.resolve_filesystem_and_path(uri)
+        args, kwargs = fsspec_filesystem_mock.call_args
+        assert args == (expected_protocol,)
+        assert kwargs == expected_kwargs
+        assert path == expected_path