ARROW-18272: [Python] Support filesystem parameter in ParquetFile (#14717)

Will fix [ARROW-18272](https://issues.apache.org/jira/browse/ARROW-18272)


Authored-by: Miles Granger <miles59923@gmail.com>
Signed-off-by: Joris Van den Bossche <jorisvandenbossche@gmail.com>

Author: milesgranger

python/pyarrow/fs.py

@@ -135,7 +135,7 @@ def _ensure_filesystem(
 
 
 def _resolve_filesystem_and_path(
-    path, filesystem=None, allow_legacy_filesystem=False
+    path, filesystem=None, allow_legacy_filesystem=False, memory_map=False
 ):
     """
     Return filesystem/path from path which could be an URI or a plain
@@ -151,7 +151,8 @@ def _resolve_filesystem_and_path(
 
     if filesystem is not None:
         filesystem = _ensure_filesystem(
-            filesystem, allow_legacy_filesystem=allow_legacy_filesystem
+            filesystem, use_mmap=memory_map,
+            allow_legacy_filesystem=allow_legacy_filesystem
         )
         if isinstance(filesystem, LocalFileSystem):
             path = _stringify_path(path)
@@ -169,7 +170,8 @@ def _resolve_filesystem_and_path(
     # if filesystem is not given, try to automatically determine one
     # first check if the file exists as a local (relative) file path
     # if not then try to parse the path as an URI
-    filesystem = LocalFileSystem()
+    filesystem = LocalFileSystem(use_mmap=memory_map)
+
     try:
         file_info = filesystem.get_file_info(path)
     except ValueError:  # ValueError means path is likely an URI

python/pyarrow/parquet/core.py

@@ -257,6 +257,10 @@ class ParquetFile:
         If not None, override the maximum total size of containers allocated
         when decoding Thrift structures. The default limit should be
         sufficient for most Parquet files.
+    filesystem : FileSystem, default None
+        If nothing passed, will be inferred based on path.
+        Path will try to be found in the local on-disk filesystem otherwise
+        it will be parsed as an URI to determine the filesystem.
 
     Examples
     --------
@@ -304,7 +308,16 @@ def __init__(self, source, *, metadata=None, common_metadata=None,
                  read_dictionary=None, memory_map=False, buffer_size=0,
                  pre_buffer=False, coerce_int96_timestamp_unit=None,
                  decryption_properties=None, thrift_string_size_limit=None,
-                 thrift_container_size_limit=None):
+                 thrift_container_size_limit=None, filesystem=None):
+
+        self._close_source = getattr(source, 'closed', True)
+
+        filesystem, source = _resolve_filesystem_and_path(
+            source, filesystem, memory_map)
+        if filesystem is not None:
+            source = filesystem.open_input_file(source)
+            self._close_source = True  # We opened it here, ensure we close it.
+
         self.reader = ParquetReader()
         self.reader.open(
             source, use_memory_map=memory_map,
@@ -315,7 +328,6 @@ def __init__(self, source, *, metadata=None, common_metadata=None,
             thrift_string_size_limit=thrift_string_size_limit,
             thrift_container_size_limit=thrift_container_size_limit,
         )
-        self._close_source = getattr(source, 'closed', True)
         self.common_metadata = common_metadata
         self._nested_paths_by_prefix = self._build_nested_paths()
 

python/pyarrow/tests/parquet/test_parquet_file.py

@@ -17,6 +17,7 @@
 
 import io
 import os
+import sys
 from unittest import mock
 
 import pytest
@@ -177,8 +178,12 @@ def test_parquet_file_pass_directory_instead_of_file(tempdir):
     path = tempdir / 'directory'
     os.mkdir(str(path))
 
-    with pytest.raises(IOError, match="Expected file path"):
+    msg = f"Cannot open for reading: path '{str(path)}' is a directory"
+    with pytest.raises(IOError) as exc:
         pq.ParquetFile(path)
+    if exc.errisinstance(PermissionError) and sys.platform == 'win32':
+        return  # Windows CI can get a PermissionError here.
+    exc.match(msg)
 
 
 def test_read_column_invalid_index():
@@ -329,3 +334,26 @@ def test_parquet_file_explicitly_closed(tempdir):
         p.read()
         assert not p.closed
     assert p.closed  # parquet file obj reports as closed
+
+
+@pytest.mark.s3
+@pytest.mark.parametrize("use_uri", (True, False))
+def test_parquet_file_with_filesystem(tempdir, s3_example_fs, use_uri):
+    s3_fs, s3_uri, s3_path = s3_example_fs
+
+    args = (s3_uri if use_uri else s3_path,)
+    kwargs = {} if use_uri else dict(filesystem=s3_fs)
+
+    table = pa.table({"a": range(10)})
+    pq.write_table(table, s3_path, filesystem=s3_fs)
+
+    parquet_file = pq.ParquetFile(*args, **kwargs)
+    assert parquet_file.read() == table
+    assert not parquet_file.closed
+    parquet_file.close()
+    assert parquet_file.closed
+
+    with pq.ParquetFile(*args, **kwargs) as f:
+        assert f.read() == table
+        assert not f.closed
+    assert f.closed