Repository: modin-project/modin
Pull request: Fix Parquet reader for partitioned files
<!--
Thank you for your contribution!
-->

## What do these changes do?
* Move `ParquetDataset` functionality to experimental
* Filter columns read in the partitions when there is at least one partition column
<!-- Please give a short brief about these changes. -->

## Related issue number
* Resolves #642 
<!-- Are there any issues opened that will be resolved by merging this change? -->

- [x] passes `flake8 modin`
- [x] passes `black --check modin`
- [ ] tests added and passing


Problem statement:
Running query over a dataframe read by read_parquet errors out with Modin 0.5.1
<!--
General questions should be asked on the mailing list modin-dev@googlegroups.com.

Before submitting an issue, please fill out the following form.
-->

### System information
- **OS Platform and Distribution (e.g., Linux Ubuntu 16.04)**: Ubuntu 19.04
- **Modin installed from (source or binary)**: pip install modin
- **Modin version**: 0.5.1
- **Python version**: 3.7.3
- **Exact command to reproduce**:
final_df = pd.read_parquet(folder, columns=fields, filters=filters or None)
                .query(query_str)
Where the query_str was of the form ("col1 == ['bar']").
<!--
You can obtain the Modin version with

python -c "import modin; print(modin.__version__)"
-->

### Describe the problem
<!-- Describe the problem clearly here. -->
Executing the pandas code snippet errors out with the traceback shown below. Needless to say, Pandas has no error. This issue is caused with Modin 0.5.1 and ray 0.7.

### 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. -->
```
File “/home/ddutt/work/a/engines/modin/engine.py”, line 91, in get_table_df
.query(query_str)
File “/home/ddutt/.local/share/virtualenvs/a-cFmMv4Vf/lib/python3.7/site-packages/modin/pandas/dataframe.py”, line 1260, in query
new_query_compiler = self._query_compiler.query(expr, **kwargs)
File “/home/ddutt/.local/share/virtualenvs/a-cFmMv4Vf/lib/python3.7/site-packages/modin/backends/pandas/query_compiler.py”, line 1669, in query
new_index = self.compute_index(0, new_data, True)
File “/home/ddutt/.local/share/virtualenvs/a-cFmMv4Vf/lib/python3.7/site-packages/modin/backends/pandas/query_compiler.py”, line 93, in compute_index
old_blocks=old_blocks,
File “/home/ddutt/.local/share/virtualenvs/a-cFmMv4Vf/lib/python3.7/site-packages/modin/engines/base/frame/partition_manager.py”, line 535, in get_indices
if len(self._partitions_cache.T)
File “/home/ddutt/.local/share/virtualenvs/a-cFmMv4Vf/lib/python3.7/site-packages/modin/engines/base/frame/partition_manager.py”, line 534, in
[idx.apply(func).get() for idx in self._partitions_cache.T[0]]
File “/home/ddutt/.local/share/virtualenvs/a-cFmMv4Vf/lib/python3.7/site-packages/modin/engines/ray/pandas_on_ray/frame/partition.py”, line 34, in get
handle_ray_task_error(e)
File “/home/ddutt/.local/share/virtualenvs/a-cFmMv4Vf/lib/python3.7/site-packages/modin/engines/ray/utils.py”, line 13, in handle_ray_task_error
raise getattr(builtins, s.split(":")[0])("".join(s.split(":")[1:]))
ValueError: Length mismatch Expected axis has 28 elements, new values have 10 elements
```

Hints from issue discussion:
Thanks @ddutt, for the report!

We're looking into the issue, others have reported running into a related issue as well. We will get this fixed as soon as possible.
@ddutt Would you be able to tell me if install this way fixes your issue?

```
pip install git+https://github.com/devin-petersohn/modin/tree/issues/642
```
The fix breaks support of predicate pushdown
I tried running `pandas.read_parquet` with `filters=None`, but I get this error in pandas:

```
In [2]: df = pandas.read_parquet("big.parquet", columns=["col1", "col2"], filters=None)
---------------------------------------------------------------------------
TypeError                                 Traceback (most recent call last)
<ipython-input-2-51e112ad24f2> in <module>
----> 1 df = pandas.read_parquet("big.parquet", columns=["col1", "col2"], filters=None)

~/.local/lib/python3.6/site-packages/pandas/io/parquet.py in read_parquet(path, engine, columns, **kwargs)
    280
    281     impl = get_engine(engine)
--> 282     return impl.read(path, columns=columns, **kwargs)

~/.local/lib/python3.6/site-packages/pandas/io/parquet.py in read(self, path, columns, **kwargs)
    127         kwargs['use_pandas_metadata'] = True
    128         result = self.api.parquet.read_table(path, columns=columns,
--> 129                                              **kwargs).to_pandas()
    130         if should_close:
    131             try:

TypeError: read_table() got an unexpected keyword argument 'filters'
```
Exactly. Predicate pushdown is only supported with ParquetDataset()
https://arrow.apache.org/docs/python/api/formats.html
Also, adding use_pandas_metadata to True in read() didn't help in removing the original error.
I'm sure you know this already, but just in case. The fix to use ParquetDataSet along with @williamma12's patches worked in 0.5.0. Something else has changed that's causing this issue.
So in order to support ~this~ predicate pushdown, we should add an `experimental` API. Anything that deviates from the pandas API should go there because it is important that users understand when they can move to/from Modin. If someone uses the pandas API in Modin it should also work in pandas. This is a little beside the point.

For internal use, predicate pushdown is important for the query planning use-case. We should definitely fix it, but it will have to go through the experimental API because we don't want it to be impossible to go to/from pandas.

Related questions to the bug: 
- Does a print before the query work? It may be an error during reading.
- Does the vectorized version `df[df["col1"] == "['bar']"]` work for these?
Also, will it work if you remove `filter` from the `read_parquet` args?

Commit message: fix conflict

Patch to review:
diff --git a/modin/engines/ray/generic/io.py b/modin/engines/ray/generic/io.py
index 9ea92769ea7..8d1faf0463a 100644
--- a/modin/engines/ray/generic/io.py
+++ b/modin/engines/ray/generic/io.py
@@ -157,9 +157,6 @@ def read_parquet(cls, path, engine, columns, **kwargs):
             for (root, dir_names, files) in os.walk(path):
                 if dir_names:
                     partitioned_columns.add(dir_names[0].split("=")[0])
-                if files:
-                    file_path = os.path.join(root, files[0])
-                    break
             partitioned_columns = list(partitioned_columns)
         else:
             directory = False
@@ -168,7 +165,7 @@ def read_parquet(cls, path, engine, columns, **kwargs):
             if directory:
                 # Path of the sample file that we will read to get the remaining
                 # columns.
-                pd = ParquetDataset(file_path)
+                pd = ParquetDataset(path)
                 column_names = pd.schema.names
             else:
                 pf = ParquetFile(path)
@@ -180,7 +177,7 @@ def read_parquet(cls, path, engine, columns, **kwargs):
         # ensure that when we do the math for the blocks, the partition column
         # will be read in along with a non partition column.
         if columns and directory and any(col in partitioned_columns for col in columns):
-            # partitioned_columns = [col for col in columns if col in partitioned_columns]
+            partitioned_columns = [col for col in columns if col in partitioned_columns]
             columns = [col for col in columns if col not in partitioned_columns]
             # If all of the columns wanted are partition columns, return an
             # empty dataframe with the desired columns.
diff --git a/modin/engines/ray/pandas_on_ray/io.py b/modin/engines/ray/pandas_on_ray/io.py
index 5232a73a8c7..cb2dbf545e7 100644
--- a/modin/engines/ray/pandas_on_ray/io.py
+++ b/modin/engines/ray/pandas_on_ray/io.py
@@ -51,8 +51,9 @@ def _read_parquet_columns(path, columns, num_splits, kwargs):  # pragma: no cove
     """
     import pyarrow.parquet as pq
 
-    df = pq.ParquetDataset(path, **kwargs).read(columns=columns).to_pandas()
-
+    kwargs["use_pandas_metadata"] = True
+    df = pq.read_table(path, columns=columns, **kwargs).to_pandas()
+    df = df[columns]
     # Append the length of the index here to build it externally
     return _split_result_for_readers(0, num_splits, df) + [len(df.index)]
 
diff --git a/modin/experimental/engines/pandas_on_ray/io_exp.py b/modin/experimental/engines/pandas_on_ray/io_exp.py
index 2b204862487..d20dc5edaca 100644
--- a/modin/experimental/engines/pandas_on_ray/io_exp.py
+++ b/modin/experimental/engines/pandas_on_ray/io_exp.py
@@ -7,7 +7,39 @@
 from modin.engines.ray.pandas_on_ray.frame.partition import PandasOnRayFramePartition
 
 
+@ray.remote
+def _read_parquet_columns(path, columns, num_splits, kwargs):  # pragma: no cover
+    """Use a Ray task to read columns from Parquet into a Pandas DataFrame.
+
+    Note: Ray functions are not detected by codecov (thus pragma: no cover)
+
+    Args:
+        path: The path of the Parquet file.
+        columns: The list of column names to read.
+        num_splits: The number of partitions to split the column into.
+
+    Returns:
+         A list containing the split Pandas DataFrames and the Index as the last
+            element. If there is not `index_col` set, then we just return the length.
+            This is used to determine the total length of the DataFrame to build a
+            default Index.
+    """
+    import pyarrow.parquet as pq
+
+    df = (
+        pq.ParquetDataset(path, **kwargs)
+        .read(columns=columns, use_pandas_metadata=True)
+        .to_pandas()
+    )
+    df = df[columns]
+    # Append the length of the index here to build it externally
+    return _split_result_for_readers(0, num_splits, df) + [len(df.index)]
+
+
 class ExperimentalPandasOnRayIO(PandasOnRayIO):
+
+    read_parquet_remote_task = _read_parquet_columns
+
     @classmethod
     def read_sql(
         cls,


Code review comments:
diff_hunk: @@ -157,9 +157,6 @@ def read_parquet(cls, path, engine, columns, **kwargs):
             for (root, dir_names, files) in os.walk(path):
                 if dir_names:
                     partitioned_columns.add(dir_names[0].split("=")[0])
-                if files:
-                    file_path = os.path.join(root, files[0])
-                    break
path: modin/engines/ray/generic/io.py
text: @user1:
We shouldn't have to continue searching for partitioned_column names once we hit a directory with a file since `os.walk` is a dfs search

diff_hunk: @@ -180,7 +177,7 @@ def read_parquet(cls, path, engine, columns, **kwargs):
         # ensure that when we do the math for the blocks, the partition column
         # will be read in along with a non partition column.
         if columns and directory and any(col in partitioned_columns for col in columns):
-            # partitioned_columns = [col for col in columns if col in partitioned_columns]
+            partitioned_columns = [col for col in columns if col in partitioned_columns]
path: modin/engines/ray/generic/io.py
text: @user1:
Won't this make it so that if we want columns 1 and 5 with the dataset partitioned on 1, 2, and 5, we will get 1 and 5 while pandas returns 1, 2, and 5?