Repository: modin-project/modin
Pull request: FIX-#4782: Exclude certain non-parquet files in `read_parquet`
Signed-off-by: Karthik Velayutham <vkarthik@ponder.io>

<!--
Thank you for your contribution!
Please review the contributing docs: https://modin.readthedocs.io/en/latest/development/contributing.html
if you have questions about contributing.
-->

## What do these changes do?

This PR fixes a bug where `read_parquet` would not ignore certain metadata files that are normally ignored by `pyarrow`.

- [x] commit message follows format outlined [here](https://modin.readthedocs.io/en/latest/development/contributing.html#commit-message-formatting)
- [x] passes `flake8 modin/ asv_bench/benchmarks scripts/doc_checker.py`
- [x] passes `black --check modin/ asv_bench/benchmarks scripts/doc_checker.py`
- [x] signed commit with `git commit -s` <!-- you can amend your commit with a signature via `git commit -amend -s` -->
- [x] Resolves #4782 
- [x] tests added and passing
- [x] module layout described at `docs/development/architecture.rst` is up-to-date <!-- if you have added, renamed or removed files or directories please update the documentation accordingly -->
- [x] added (Issue Number: PR title (PR Number)) and github username to release notes for next major release <!-- e.g. DOCS-#4077: Add release notes template to docs folder (#4078) -->


Problem statement:
`read_parquet` fails when there are other non-parquet metadata files in folder
### System information
- **OS Platform and Distribution (e.g., Linux Ubuntu 16.04)**: macOS 12.2.1
- **Modin version** (`modin.__version__`): 0.15.0+67.g5a3eea4b
- **Python version**: 3.9.12
- **Code we can use to reproduce**:
```python
In [1]: import modin.pandas as pd

In [2]: import pandas

In [3]: import os

In [4]: df = pd.DataFrame(
   ...:         {
   ...:             "foo": ["one", "one", "one", "two", "two", "two"],
   ...:             "bar": ["A", "B", "C", "A", "B", "C"],
   ...:             "baz": [1, 2, 3, 4, 5, 6],
   ...:             "zoo": ["x", "y", "z", "q", "w", "t"],
   ...:         }
   ...:     )
UserWarning: Ray execution environment not yet initialized. Initializing...
To remove this warning, run the following python code before doing dataframe operations:

    import ray
    ray.init(runtime_env={'env_vars': {'__MODIN_AUTOIMPORT_PANDAS__': '1'}})

UserWarning: On Macs, Ray's performance is known to degrade with object store size greater than 2.0 GiB. Ray by default does not allow setting an object store size greater than that. Modin is overriding that default limit because it would rather have a larger, slower object store than spill to disk more often. To override Modin's behavior, you can initialize Ray yourself.
UserWarning: Distributing <class 'dict'> object. This may take some time.

In [5]: pdf = df._to_pandas()

In [6]: os.makedirs("testing")

In [7]: df.to_parquet("testing/")

In [9]: os.listdir("testing/")
Out[9]: ['_committed_blah', 'part-0000.snappy.parquet']

In [10]: df = pd.read_parquet("testing/")
---------------------------------------------------------------------------
ArrowInvalid                              Traceback (most recent call last)
Input In [10], in <cell line: 1>()
----> 1 df = pd.read_parquet("testing/")
...
File ~/opt/anaconda3/envs/modin/lib/python3.9/site-packages/pyarrow/parquet/__init__.py:277, in ParquetFile.__init__(self, source, metadata, common_metadata, read_dictionary, memory_map, buffer_size, pre_buffer, coerce_int96_timestamp_unit, decryption_properties)
    272 def __init__(self, source, metadata=None, common_metadata=None,
    273              read_dictionary=None, memory_map=False, buffer_size=0,
    274              pre_buffer=False, coerce_int96_timestamp_unit=None,
    275              decryption_properties=None):
    276     self.reader = ParquetReader()
--> 277     self.reader.open(
    278         source, use_memory_map=memory_map,
    279         buffer_size=buffer_size, pre_buffer=pre_buffer,
    280         read_dictionary=read_dictionary, metadata=metadata,
    281         coerce_int96_timestamp_unit=coerce_int96_timestamp_unit,
    282         decryption_properties=decryption_properties
    283     )
    284     self.common_metadata = common_metadata
    285     self._nested_paths_by_prefix = self._build_nested_paths()

File ~/opt/anaconda3/envs/modin/lib/python3.9/site-packages/pyarrow/_parquet.pyx:1213, in pyarrow._parquet.ParquetReader.open()

File ~/opt/anaconda3/envs/modin/lib/python3.9/site-packages/pyarrow/error.pxi:100, in pyarrow.lib.check_status()

ArrowInvalid: Parquet magic bytes not found in footer. Either the file is corrupted or this is not a parquet file.

In [11]: pdf = pandas.read_parquet("testing/")

In [12]: pdf
Out[12]: 
   foo bar  baz zoo
0  one   A    1   x
1  one   B    2   y
2  one   C    3   z
3  two   A    4   q
4  two   B    5   w
5  two   C    6   t

In [13]: 

```
### Describe the problem
Currently `read_parquet` will fail if we try reading a directory that contains certain metadata files that are normally ignored by PyArrow's default parquet reader. For example, writing out parquet files with Spark will leave some files that start with `_committed_xxx` or `_started_xxx` which is usually some lingering metadata from DBIO. Our current implementation of `read_parquet` tries to read these metadata files as parquet files instead of ignoring them like PyArrow. We should mimic pandas/pyarrow behavior in this case. 

### Source code / logs
<!-- Include any logs or source code that would be helpful to diagnose the problem. If including tracebacks, please include the full traceback. Large logs and files should be attached. Try to provide a reproducible test case that is the bare minimum necessary to generate the problem. -->

<details>
<summary>Show stack trace</summary>

```
----> 1 df = pd.read_parquet('blah/')

File ~/Documents/modin/modin/logging/logger_decorator.py:128, in enable_logging.<locals>.decorator.<locals>.run_and_log(*args, **kwargs)
    113 """
    114 Compute function with logging if Modin logging is enabled.
    115 
   (...)
    125 Any
    126 """
    127 if LogMode.get() == "disable":
--> 128     return obj(*args, **kwargs)
    130 logger = get_logger()
    131 logger_level = getattr(logger, log_level)

File ~/Documents/modin/modin/_compat/pandas_api/latest/io.py:209, in read_parquet(path, engine, columns, storage_options, use_nullable_dtypes, **kwargs)
    205 Engine.subscribe(_update_engine)
    206 from modin.core.execution.dispatching.factories.dispatcher import FactoryDispatcher
    208 return DataFrame(
--> 209     query_compiler=FactoryDispatcher.read_parquet(
    210         path=path,
    211         engine=engine,
    212         columns=columns,
    213         storage_options=storage_options,
    214         use_nullable_dtypes=use_nullable_dtypes,
    215         **kwargs,
    216     )
    217 )

File ~/Documents/modin/modin/core/execution/dispatching/factories/dispatcher.py:180, in FactoryDispatcher.read_parquet(cls, **kwargs)
    177 @classmethod
    178 @_inherit_docstrings(factories.BaseFactory._read_parquet)
    179 def read_parquet(cls, **kwargs):
--> 180     return cls.__factory._read_parquet(**kwargs)

File ~/Documents/modin/modin/core/execution/dispatching/factories/factories.py:207, in BaseFactory._read_parquet(cls, **kwargs)
    199 @classmethod
    200 @doc(
    201     _doc_io_method_template,
   (...)
    205 )
    206 def _read_parquet(cls, **kwargs):
--> 207     return cls.io_cls.read_parquet(**kwargs)

File ~/Documents/modin/modin/logging/logger_decorator.py:128, in enable_logging.<locals>.decorator.<locals>.run_and_log(*args, **kwargs)
    113 """
    114 Compute function with logging if Modin logging is enabled.
    115 
   (...)
    125 Any
    126 """
    127 if LogMode.get() == "disable":
--> 128     return obj(*args, **kwargs)
    130 logger = get_logger()
    131 logger_level = getattr(logger, log_level)

File ~/Documents/modin/modin/core/io/file_dispatcher.py:154, in FileDispatcher.read(cls, *args, **kwargs)
    131 @classmethod
    132 def read(cls, *args, **kwargs):
    133     """
    134     Read data according passed `args` and `kwargs`.
    135 
   (...)
    152     postprocessing work on the resulting query_compiler object.
    153     """
--> 154     query_compiler = cls._read(*args, **kwargs)
    155     # TODO (devin-petersohn): Make this section more general for non-pandas kernel
    156     # implementations.
    157     if StorageFormat.get() == "Pandas":

File ~/Documents/modin/modin/logging/logger_decorator.py:128, in enable_logging.<locals>.decorator.<locals>.run_and_log(*args, **kwargs)
    113 """
    114 Compute function with logging if Modin logging is enabled.
    115 
   (...)
    125 Any
    126 """
    127 if LogMode.get() == "disable":
--> 128     return obj(*args, **kwargs)
    130 logger = get_logger()
    131 logger_level = getattr(logger, log_level)

File ~/Documents/modin/modin/core/io/column_stores/parquet_dispatcher.py:432, in ParquetDispatcher._read(cls, path, engine, columns, **kwargs)
    425 column_names = dataset.schema.names if not columns else columns
    426 columns = [
    427     c
    428     for c in column_names
    429     if c not in index_columns and not PQ_INDEX_REGEX.match(c)
    430 ]
--> 432 return cls.build_query_compiler(path, columns, index_columns, **kwargs)

File ~/Documents/modin/modin/logging/logger_decorator.py:128, in enable_logging.<locals>.decorator.<locals>.run_and_log(*args, **kwargs)
    113 """
    114 Compute function with logging if Modin logging is enabled.
    115 
   (...)
    125 Any
    126 """
    127 if LogMode.get() == "disable":
--> 128     return obj(*args, **kwargs)
    130 logger = get_logger()
    131 logger_level = getattr(logger, log_level)

File ~/Documents/modin/modin/core/io/column_stores/parquet_dispatcher.py:339, in ParquetDispatcher.build_query_compiler(cls, path, columns, index_columns, **kwargs)
    337 storage_options = kwargs.pop("storage_options", {}) or {}
    338 col_partitions, column_widths = cls.build_columns(columns)
--> 339 partition_ids = cls.call_deploy(path, col_partitions, storage_options, **kwargs)
    340 index, sync_index = cls.build_index(
    341     path, partition_ids, index_columns, storage_options
    342 )
    343 remote_parts = cls.build_partition(partition_ids, column_widths)

File ~/Documents/modin/modin/logging/logger_decorator.py:128, in enable_logging.<locals>.decorator.<locals>.run_and_log(*args, **kwargs)
    113 """
    114 Compute function with logging if Modin logging is enabled.
    115 
   (...)
    125 Any
    126 """
    127 if LogMode.get() == "disable":
--> 128     return obj(*args, **kwargs)
    130 logger = get_logger()
    131 logger_level = getattr(logger, log_level)

File ~/Documents/modin/modin/core/io/column_stores/parquet_dispatcher.py:146, in ParquetDispatcher.call_deploy(cls, fname, col_partitions, storage_options, **kwargs)
    144 for file in parquet_files:
    145     with filesystem.open(file) as f:
--> 146         row_groups = ParquetFile(f).num_row_groups
    147         row_groups_per_file.append(row_groups)
    148         num_row_groups += row_groups

File ~/opt/anaconda3/envs/modin/lib/python3.9/site-packages/pyarrow/parquet/__init__.py:277, in ParquetFile.__init__(self, source, metadata, common_metadata, read_dictionary, memory_map, buffer_size, pre_buffer, coerce_int96_timestamp_unit, decryption_properties)
    272 def __init__(self, source, metadata=None, common_metadata=None,
    273              read_dictionary=None, memory_map=False, buffer_size=0,
    274              pre_buffer=False, coerce_int96_timestamp_unit=None,
    275              decryption_properties=None):
    276     self.reader = ParquetReader()
--> 277     self.reader.open(
    278         source, use_memory_map=memory_map,
    279         buffer_size=buffer_size, pre_buffer=pre_buffer,
    280         read_dictionary=read_dictionary, metadata=metadata,
    281         coerce_int96_timestamp_unit=coerce_int96_timestamp_unit,
    282         decryption_properties=decryption_properties
    283     )
    284     self.common_metadata = common_metadata
    285     self._nested_paths_by_prefix = self._build_nested_paths()

File ~/opt/anaconda3/envs/modin/lib/python3.9/site-packages/pyarrow/_parquet.pyx:1213, in pyarrow._parquet.ParquetReader.open()

File ~/opt/anaconda3/envs/modin/lib/python3.9/site-packages/pyarrow/error.pxi:100, in pyarrow.lib.check_status()

ArrowInvalid: Parquet magic bytes not found in footer. Either the file is corrupted or this is not a parquet file.
```

</details>

Hints from issue discussion:

Commit message: Add release notes and minor doc fix

Patch to review:
diff --git a/docs/release_notes/release_notes-0.16.0.rst b/docs/release_notes/release_notes-0.16.0.rst
index 1b6fdaab1f3..852f4ea0c9b 100644
--- a/docs/release_notes/release_notes-0.16.0.rst
+++ b/docs/release_notes/release_notes-0.16.0.rst
@@ -29,7 +29,8 @@ Key Features and Updates
   * FIX-#4652: Support categorical data in `from_dataframe` (#4737)
   * FIX-#4756: Correctly propagate `storage_options` in `read_parquet` (#4764)
   * FIX-#4657: Use `fsspec` for handling s3/http-like paths instead of `s3fs` (#4710)
-  * FIX-#4676: drain sub-virtual-partition call queues (#4695)    
+  * FIX-#4676: drain sub-virtual-partition call queues (#4695)
+  * FIX-#4782: Exclude certain non-parquet files in `read_parquet` (#4783)    
 * Performance enhancements
   * PERF-#4182: Add cell-wise execution for binary ops, fix bin ops for empty dataframes (#4391)
   * PERF-#4288: Improve perf of `groupby.mean` for narrow data (#4591)
diff --git a/modin/core/io/column_stores/parquet_dispatcher.py b/modin/core/io/column_stores/parquet_dispatcher.py
index 135c34cf546..8a66298bd99 100644
--- a/modin/core/io/column_stores/parquet_dispatcher.py
+++ b/modin/core/io/column_stores/parquet_dispatcher.py
@@ -35,7 +35,7 @@ class ParquetDispatcher(ColumnStoreDispatcher):
     """Class handles utils for reading `.parquet` files."""
 
     @classmethod
-    def get_fsspec_files(cls, path, storage_options):
+    def get_fs_and_files(cls, path, storage_options):
         """
         Retrieve filesystem interface and list of files from path.
 
@@ -65,22 +65,39 @@ def _unstrip_protocol(protocol, path):
 
         if isinstance(path, AbstractBufferedFile):
             return path.fs, [path]
-        protocol, path = split_protocol(path)
-        filesystem = get_filesystem_class(protocol)(**storage_options)
-        if filesystem.stat(path)["type"] == "directory":
-            files = []
-            for path in sorted(filesystem.find(path)):
-                if version.parse(fsspec.__version__) < version.parse("2022.5.0"):
-                    files.append(_unstrip_protocol(filesystem.protocol, path))
-                else:
-                    files.append(filesystem.unstrip_protocol(path))
-            return filesystem, files
-
-        return filesystem, [
-            _unstrip_protocol(filesystem.protocol, path)
-            if version.parse(fsspec.__version__) < version.parse("2022.5.0")
-            else filesystem.unstrip_protocol(path)
-        ]
+        filesystem = get_filesystem_class(split_protocol(path)[0])(**storage_options)
+        dataset = cls._get_dataset(path, storage_options)
+        files = []
+        for fpath in dataset.files:
+            if version.parse(fsspec.__version__) < version.parse("2022.5.0"):
+                files.append(_unstrip_protocol(filesystem.protocol, fpath))
+            else:
+                files.append(filesystem.unstrip_protocol(fpath))
+
+        return filesystem, files
+
+    @classmethod
+    def _get_dataset(cls, path, storage_options):
+        """
+        Retrieve parquet dataset for given path.
+
+        Parameters
+        ----------
+        path : str, path object or file-like object
+            Path to dataset.
+        storage_options : dict
+            Parameters for specific storage engine.
+
+        Returns
+        -------
+        dataset : ParquetDataset
+            Dataset object of given Parquet file-path.
+        """
+        from pyarrow.parquet import ParquetDataset
+
+        fs, fs_path = cls._get_fs_and_fs_path(path, storage_options)
+        dataset = ParquetDataset(fs_path, filesystem=fs, use_legacy_dataset=False)
+        return dataset
 
     @classmethod
     def _get_fs_and_fs_path(cls, path, storage_options):
@@ -106,13 +123,13 @@ def _get_fs_and_fs_path(cls, path, storage_options):
         )
 
     @classmethod
-    def call_deploy(cls, fname, col_partitions, storage_options, **kwargs):
+    def call_deploy(cls, path, col_partitions, storage_options, **kwargs):
         """
         Deploy remote tasks to the workers with passed parameters.
 
         Parameters
         ----------
-        fname : str, path object or file-like object
+        path : str, path object or file-like object
             Name of the file to read.
         col_partitions : list
             List of arrays with columns names that should be read
@@ -135,7 +152,7 @@ def call_deploy(cls, fname, col_partitions, storage_options, **kwargs):
         if len(col_partitions) == 0:
             return []
 
-        filesystem, parquet_files = cls.get_fsspec_files(fname, storage_options)
+        filesystem, parquet_files = cls.get_fs_and_files(path, storage_options)
 
         row_groups_per_file = []
         num_row_groups = 0
@@ -387,7 +404,6 @@ def _read(cls, path, engine, columns, **kwargs):
             "pyarrow",
             "pyarrow is required to read parquet files.",
         )
-        from pyarrow.parquet import ParquetDataset
         from modin.pandas.io import PQ_INDEX_REGEX
 
         if isinstance(path, str) and os.path.isdir(path):
@@ -413,9 +429,7 @@ def _read(cls, path, engine, columns, **kwargs):
                     reason="Mixed partitioning columns in Parquet",
                     **kwargs,
                 )
-        # url_to_fs returns the fs and path formatted for the specific fs
-        fs, fs_path = cls._get_fs_and_fs_path(path, kwargs.get("storage_options") or {})
-        dataset = ParquetDataset(fs_path, filesystem=fs, use_legacy_dataset=False)
+        dataset = cls._get_dataset(path, kwargs.get("storage_options") or {})
         index_columns = (
             dataset.schema.pandas_metadata.get("index_columns", [])
             if dataset.schema.pandas_metadata
diff --git a/modin/pandas/test/test_io.py b/modin/pandas/test/test_io.py
index 2174ca3ccc9..2ca66b96105 100644
--- a/modin/pandas/test/test_io.py
+++ b/modin/pandas/test/test_io.py
@@ -1403,6 +1403,14 @@ def test_read_parquet_directory(
             )
             start_row = end_row
         path = make_parquet_dir(dfs_by_filename, row_group_size)
+
+        # There are specific files that PyArrow will try to ignore by default
+        # in a parquet directory. One example are files that start with '_'. Our
+        # previous implementation tried to read all files in a parquet directory,
+        # but we now make use of PyArrow to ensure the directory is valid.
+        with open(os.path.join(path, "_committed_file"), "w+") as f:
+            f.write("testingtesting")
+
         eval_io(
             fn_name="read_parquet",
             # read_parquet kwargs


Code review comments:
diff_hunk: @@ -65,22 +65,39 @@ def _unstrip_protocol(protocol, path):
 
         if isinstance(path, AbstractBufferedFile):
             return path.fs, [path]
-        protocol, path = split_protocol(path)
-        filesystem = get_filesystem_class(protocol)(**storage_options)
-        if filesystem.stat(path)["type"] == "directory":
-            files = []
-            for path in sorted(filesystem.find(path)):
-                if version.parse(fsspec.__version__) < version.parse("2022.5.0"):
-                    files.append(_unstrip_protocol(filesystem.protocol, path))
-                else:
-                    files.append(filesystem.unstrip_protocol(path))
-            return filesystem, files
-
-        return filesystem, [
-            _unstrip_protocol(filesystem.protocol, path)
-            if version.parse(fsspec.__version__) < version.parse("2022.5.0")
-            else filesystem.unstrip_protocol(path)
-        ]
+        filesystem = get_filesystem_class(split_protocol(path)[0])(**storage_options)
+        dataset = cls._get_dataset(path, storage_options)
+        files = []
+        for fpath in dataset.files:
+            if version.parse(fsspec.__version__) < version.parse("2022.5.0"):
path: modin/core/io/column_stores/parquet_dispatcher.py
text: @user1:
please swap `fsspec` version check and `for` loop - it's better to do two loops, one in `if` and other in `else`, this would work faster

@author:
Oh right this is a good point. I looked into the implementation of parse and it's a ton of regex matching which is usually pretty expensive.  Thanks for the suggestion!