Support viewfs scheme along side with hdfs (#777)

Author: yothinix

pyiceberg/io/__init__.py

@@ -284,6 +284,7 @@ def delete(self, location: Union[str, InputFile, OutputFile]) -> None:
     "gs": [ARROW_FILE_IO],
     "file": [ARROW_FILE_IO, FSSPEC_FILE_IO],
     "hdfs": [ARROW_FILE_IO],
+    "viewfs": [ARROW_FILE_IO],
     "abfs": [FSSPEC_FILE_IO],
     "abfss": [FSSPEC_FILE_IO],
 }

pyiceberg/io/pyarrow.py

@@ -332,7 +332,7 @@ def parse_location(location: str) -> Tuple[str, str, str]:
         uri = urlparse(location)
         if not uri.scheme:
             return "file", uri.netloc, os.path.abspath(location)
-        elif uri.scheme == "hdfs":
+        elif uri.scheme in ("hdfs", "viewfs"):
             return uri.scheme, uri.netloc, uri.path
         else:
             return uri.scheme, uri.netloc, f"{uri.netloc}{uri.path}"
@@ -356,12 +356,12 @@ def _initialize_fs(self, scheme: str, netloc: Optional[str] = None) -> FileSyste
                 client_kwargs["connect_timeout"] = float(connect_timeout)
 
             return S3FileSystem(**client_kwargs)
-        elif scheme == "hdfs":
+        elif scheme in ("hdfs", "viewfs"):
             from pyarrow.fs import HadoopFileSystem
 
             hdfs_kwargs: Dict[str, Any] = {}
             if netloc:
-                return HadoopFileSystem.from_uri(f"hdfs://{netloc}")
+                return HadoopFileSystem.from_uri(f"{scheme}://{netloc}")
             if host := self.properties.get(HDFS_HOST):
                 hdfs_kwargs["host"] = host
             if port := self.properties.get(HDFS_PORT):