Use pyarrow S3 file system at read time for arrow parquet engine

dask/dataframe/io/parquet/arrow.py

@@ -335,9 +335,15 @@ def read_metadata(
         ignore_metadata_file=False,
         metadata_task_size=0,
         parquet_file_extension=None,
+        storage_options=None,
         **kwargs,
     ):
 
+        # Set default open_file_options for remote filesystems
+        kwargs["open_file_options"] = cls._default_open_file_options(
+            fs, storage_options, kwargs.pop("open_file_options", {})
+        )
+
         # Stage 1: Collect general dataset information
         dataset_info = cls._collect_dataset_info(
             paths,
@@ -762,6 +768,30 @@ def write_metadata(cls, parts, meta, fs, path, append=False, **kwargs):
     # Private Class Methods
     #
 
+    @classmethod
+    def _default_open_file_options(cls, fs, storage_options, input_options):
+        """Set default open_file_options
+
+        Stick with user-provided options (if there are any). Otherwise:
+        Use native pyarrow filesystem for 'open_file_func' if possible.
+
+        Currently supports ``s3fs`` -> ``pyarrow.fs.S3FileSystem``.
+        """
+        if "s3" in fs.protocol and not input_options:
+            pa_option_map = {"anon": "anonymous"}
+            try:
+                from pyarrow import fs as pa_fs
+
+                pa_options = {pa_option_map[k]: v for k, v in storage_options.items()}
+                return {
+                    "open_file_func": pa_fs.S3FileSystem(**pa_options).open_input_file
+                }
+            except KeyError:
+                # Could not map one or more ``storage_options``
+                # keys to ``S3FileSystem`` options
+                pass
+        return input_options
+
     @classmethod
     def _collect_dataset_info(
         cls,

dask/dataframe/io/parquet/core.py

@@ -483,6 +483,7 @@ def read_parquet(
         ignore_metadata_file=ignore_metadata_file,
         metadata_task_size=metadata_task_size,
         parquet_file_extension=parquet_file_extension,
+        storage_options=storage_options,
         **kwargs,
     )
 

dask/dataframe/io/parquet/fastparquet.py

@@ -828,6 +828,7 @@ def read_metadata(
         ignore_metadata_file=False,
         metadata_task_size=None,
         parquet_file_extension=None,
+        storage_options=None,
         **kwargs,
     ):