Repository: dask/dask
Pull request: Fix "file_path_0" bug in aggregate_row_groups
Closes #5624 

PR #5607 does not check if there is a `"file_path_0"` entry in the `statistics` item before checking it's value. This entry will **not** be present in the case that each partition corresponds to a distinct file (because there are not multiple row-groups per file).  The changes here just add a simple check, and a corresponding test.  

- [ ] Tests added / passed
- [ ] Passes `black dask` / `flake8 dask`


Problem statement:
Support chunksize parameter for read_parquet with a single file
I'd like to be able to read a single parquet file into multiple partitions, determined by the chunksize.

Without chunksize
```python
import pandas as pd
import dask.dataframe as dd
​
df = pd.DataFrame({"a":range(100000), "b":range(100000)})
df.to_parquet("out.parquet")
​
df = dd.read_parquet("out.parquet",
                     gather_statistics=True, split_row_groups=True)
df.npartitions
1
```

With chunksize
```python
import pandas as pd
import dask.dataframe as dd

df = pd.DataFrame({"a":range(100000), "b":range(100000)})
df.to_parquet("out.parquet")

df = dd.read_parquet("out.parquet",
                     chunksize="10 MiB",
                     gather_statistics=True, split_row_groups=True)
---------------------------------------------------------------------------
KeyError                                  Traceback (most recent call last)
<ipython-input-20-25c43bb02cd0> in <module>
      7 df = dd.read_parquet("out.parquet",
      8                      chunksize="10 MiB",
----> 9                      gather_statistics=True, split_row_groups=True)

/opt/conda/envs/rapids/lib/python3.7/site-packages/dask/dataframe/io/parquet/core.py in read_parquet(path, columns, filters, categories, index, storage_options, engine, gather_statistics, split_row_groups, chunksize, **kwargs)
    229     # Parse dataset statistics from metadata (if available)
    230     parts, divisions, index, index_in_columns = process_statistics(
--> 231         parts, statistics, filters, index, chunksize
    232     )
    233 

/opt/conda/envs/rapids/lib/python3.7/site-packages/dask/dataframe/io/parquet/core.py in process_statistics(parts, statistics, filters, index, chunksize)
    619         # Aggregate parts/statistics if we are splitting by row-group
    620         if chunksize:
--> 621             parts, statistics = aggregate_row_groups(parts, statistics, chunksize)
    622 
    623         out = sorted_columns(statistics)

/opt/conda/envs/rapids/lib/python3.7/site-packages/dask/dataframe/io/parquet/core.py in aggregate_row_groups(parts, stats, chunksize)
    722 
    723 def aggregate_row_groups(parts, stats, chunksize):
--> 724     if not stats[0]["file_path_0"]:
    725         return parts, stats
    726 

KeyError: 'file_path_0'
```

Hints from issue discussion:
cc @rjzamora because of the file_path_0 bug

@beckernick just to be clear the finest granularity that we're able to manage is an individual row group.  Parquet doesn't really support reading data at finer granularities.  In your case I would ask the question of how did `pandas.DataFrame.to_parquet` choose to save your file?  Did it use row groups?  If not then there is nothing that we can do here.
Thanks @mrocklin , makes sense 👍 . This issue was primarily motivated by the `file_path_0` bug that comes up while using `chunksize`.
Yup.  Thanks for raising!
Thanks for raising @beckernick !  Ill add a test for the case like this where the row-group aggregation is impossible.  Although row-group aggregation cannot work in this case, there should certainly not be an error - Sorry about that!

Commit message: simple fix for #5624

Patch to review:
diff --git a/dask/dataframe/io/parquet/core.py b/dask/dataframe/io/parquet/core.py
index 7cee9105cb3..9508e1c8c19 100644
--- a/dask/dataframe/io/parquet/core.py
+++ b/dask/dataframe/io/parquet/core.py
@@ -721,7 +721,7 @@ def set_index_columns(meta, index, columns, index_in_columns, auto_index_allowed
 
 
 def aggregate_row_groups(parts, stats, chunksize):
-    if not stats[0]["file_path_0"]:
+    if ("file_path_0" not in stats[0]) or (stats[0]["file_path_0"] is None):
         return parts, stats
 
     parts_agg = []
diff --git a/dask/dataframe/io/tests/test_parquet.py b/dask/dataframe/io/tests/test_parquet.py
index c43fd66d898..5a7abdbb7a7 100644
--- a/dask/dataframe/io/tests/test_parquet.py
+++ b/dask/dataframe/io/tests/test_parquet.py
@@ -2170,3 +2170,22 @@ def test_chunksize(tmpdir, chunksize, engine, metadata):
         remainder = (df_byte_size % parse_bytes(chunksize)) > 0
         expected += int(remainder) * nparts
         assert ddf2.npartitions == max(nparts, expected)
+
+
+@write_read_engines()
+def test_roundtrip_pandas_chunksize(tmpdir, write_engine, read_engine):
+    path = str(tmpdir.join("test.parquet"))
+    pdf = df.copy()
+    pdf.index.name = "index"
+    pdf.to_parquet(path, engine=write_engine)
+
+    ddf_read = dd.read_parquet(
+        path,
+        engine=read_engine,
+        chunksize="10 kiB",
+        gather_statistics=True,
+        split_row_groups=True,
+        index="index",
+    )
+
+    assert_eq(pdf, ddf_read)


Code review comments:
diff_hunk: @@ -721,7 +721,7 @@ def set_index_columns(meta, index, columns, index_in_columns, auto_index_allowed
 
 
 def aggregate_row_groups(parts, stats, chunksize):
-    if not stats[0]["file_path_0"]:
+    if ("file_path_0" not in stats[0]) or (stats[0]["file_path_0"] is None):
path: dask/dataframe/io/parquet/core.py
text: @user1:
Perhaps the following?

```suggestion
    if not stats[0].get("file_path_0", None):
```