ARROW-16719: [Python] Add path/URI + filesystem handling to parquet.read_metadata

python/pyarrow/parquet/__init__.py

@@ -18,6 +18,7 @@
 
 from collections import defaultdict
 from concurrent import futures
+from contextlib import nullcontext
 from functools import partial, reduce
 
 import json
@@ -3388,7 +3389,8 @@ def write_metadata(schema, where, metadata_collector=None, **kwargs):
         metadata.write_metadata_file(where)
 
 
-def read_metadata(where, memory_map=False, decryption_properties=None):
+def read_metadata(where, memory_map=False, decryption_properties=None,
+                  filesystem=None):
     """
     Read FileMetaData from footer of a single Parquet file.
 
@@ -3399,6 +3401,10 @@ def read_metadata(where, memory_map=False, decryption_properties=None):
         Create memory map when the source is a file path.
     decryption_properties : FileDecryptionProperties, default None
         Decryption properties for reading encrypted 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.
 
     Returns
     -------
@@ -3421,11 +3427,19 @@ def read_metadata(where, memory_map=False, decryption_properties=None):
       format_version: 2.6
       serialized_size: 561
     """
-    return ParquetFile(where, memory_map=memory_map,
-                       decryption_properties=decryption_properties).metadata
+    filesystem, where = _resolve_filesystem_and_path(where, filesystem)
+    file_ctx = nullcontext()
+    if filesystem is not None:
+        file_ctx = where = filesystem.open_input_file(where)
+
+    with file_ctx:
+        file = ParquetFile(where, memory_map=memory_map,
+                           decryption_properties=decryption_properties)
+        return file.metadata
 
 
-def read_schema(where, memory_map=False, decryption_properties=None):
+def read_schema(where, memory_map=False, decryption_properties=None,
+                filesystem=None):
     """
     Read effective Arrow schema from Parquet file metadata.
 
@@ -3436,6 +3450,10 @@ def read_schema(where, memory_map=False, decryption_properties=None):
         Create memory map when the source is a file path.
     decryption_properties : FileDecryptionProperties, default None
         Decryption properties for reading encrypted 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.
 
     Returns
     -------
@@ -3453,6 +3471,13 @@ def read_schema(where, memory_map=False, decryption_properties=None):
     n_legs: int64
     animal: string
     """
-    return ParquetFile(
-        where, memory_map=memory_map,
-        decryption_properties=decryption_properties).schema.to_arrow_schema()
+    filesystem, where = _resolve_filesystem_and_path(where, filesystem)
+    file_ctx = nullcontext()
+    if filesystem is not None:
+        file_ctx = where = filesystem.open_input_file(where)
+
+    with file_ctx:
+        file = ParquetFile(
+            where, memory_map=memory_map,
+            decryption_properties=decryption_properties)
+        return file.schema.to_arrow_schema()

python/pyarrow/tests/parquet/test_metadata.py

@@ -18,12 +18,15 @@
 import datetime
 import decimal
 from collections import OrderedDict
+import os
 
 import numpy as np
 import pytest
 
 import pyarrow as pa
 from pyarrow.tests.parquet.common import _check_roundtrip, make_sample_file
+from pyarrow.fs import LocalFileSystem, FileSystem
+from pyarrow.tests import util
 
 try:
     import pyarrow.parquet as pq
@@ -533,6 +536,44 @@ def test_metadata_exceeds_message_size():
     metadata = pq.read_metadata(pa.BufferReader(buf))
 
 
+def test_metadata_schema_filesystem(tmpdir):
+    table = pa.table({"a": [1, 2, 3]})
+
+    # URI writing to local file.
+    fname = "data.parquet"
+    file_path = 'file:///' + os.path.join(str(tmpdir), fname)
+
+    pq.write_table(table, file_path)
+
+    # Get expected `metadata` from path.
+    metadata = pq.read_metadata(tmpdir / fname)
+    schema = table.schema
+
+    assert pq.read_metadata(file_path).equals(metadata)
+    assert pq.read_metadata(
+        fname, filesystem=f'file:///{tmpdir}').equals(metadata)
+
+    assert pq.read_schema(file_path).equals(schema)
+    assert pq.read_schema(fname, filesystem=f'file:///{tmpdir}').equals(schema)
+
+    with util.change_cwd(tmpdir):
+        # Pass `filesystem` arg
+        assert pq.read_metadata(
+            fname, filesystem=LocalFileSystem()).equals(metadata)
+        assert pq.read_metadata(
+            fname, filesystem=LocalFileSystem.get_instance()).equals(metadata)
+
+        assert pq.read_schema(
+            fname, filesystem=LocalFileSystem()).equals(schema)
+        assert pq.read_schema(
+            fname, filesystem=LocalFileSystem.get_instance()).equals(schema)
+
+        err_msg = ('`filesystem` argument must be a FileSystem'
+                   ' instance or a valid file system URI')
+        with pytest.raises(TypeError, match=err_msg):
+            pq.read_metadata(fname, filesystem=FileSystem())
+
+
 def test_metadata_equals():
     table = pa.table({"a": [1, 2, 3]})
     with pa.BufferOutputStream() as out: