Repository: dask/dask

Issue:
Example outer product implementation doesn't work
The docstring for `dask.array.blockwise` contains this example:

```python
z = blockwise(operator.mul, 'ij', x, 'i', y, 'j', dtype='f8')  # doctest: +SKIP
```

However, it doesn't actually work (computing it fails), because `operator.mul` between two 1D chunks produces another 1D chunk, not a 2D outer product of the chunks. I think replacing it with `np.outer` will work, but I'll double-check and make a PR.

Discussion:

Patch:
diff --git a/dask/array/blockwise.py b/dask/array/blockwise.py
index 72fb9bc2f..80b9aad23 100644
--- a/dask/array/blockwise.py
+++ b/dask/array/blockwise.py
@@ -54,17 +54,34 @@ def blockwise(
     --------
     2D embarrassingly parallel operation from two arrays, x, and y.
 
-    >>> z = blockwise(operator.add, 'ij', x, 'ij', y, 'ij', dtype='f8')  # z = x + y  # doctest: +SKIP
-
-    Outer product multiplying x by y, two 1-d vectors
-
-    >>> z = blockwise(operator.mul, 'ij', x, 'i', y, 'j', dtype='f8')  # doctest: +SKIP
+    >>> import operator, numpy as np, dask.array as da
+    >>> x = da.from_array([[1, 2],
+    ...                    [3, 4]], chunks=(1, 2))
+    >>> y = da.from_array([[10, 20],
+    ...                    [0, 0]])
+    >>> z = blockwise(operator.add, 'ij', x, 'ij', y, 'ij', dtype='f8')
+    >>> z.compute()
+    array([[11, 22],
+           [ 3,  4]])
+
+    Outer product multiplying a by b, two 1-d vectors
+
+    >>> a = da.from_array([0, 1, 2], chunks=1)
+    >>> b = da.from_array([10, 50, 100], chunks=1)
+    >>> z = blockwise(np.outer, 'ij', a, 'i', b, 'j', dtype='f8')
+    >>> z.compute()
+    array([[ 0,    0,   0],
+           [ 10,  50, 100],
+           [ 20, 100, 200]])
 
     z = x.T
 
-    >>> z = blockwise(np.transpose, 'ji', x, 'ij', dtype=x.dtype)  # doctest: +SKIP
+    >>> z = blockwise(np.transpose, 'ji', x, 'ij', dtype=x.dtype)
+    >>> z.compute()
+    array([[1, 3],
+           [2, 4]])
 
-    The transpose case above is illustrative because it does same transposition
+    The transpose case above is illustrative because it does transposition
     both on each in-memory block by calling ``np.transpose`` and on the order
     of the blocks themselves, by switching the order of the index ``ij -> ji``.
 
@@ -73,7 +90,10 @@ def blockwise(
 
     z = X + Y.T
 
-    >>> z = blockwise(lambda x, y: x + y.T, 'ij', x, 'ij', y, 'ji', dtype='f8')  # doctest: +SKIP
+    >>> z = blockwise(lambda x, y: x + y.T, 'ij', x, 'ij', y, 'ji', dtype='f8')
+    >>> z.compute()
+    array([[11,  2],
+           [23,  4]])
 
     Any index, like ``i`` missing from the output index is interpreted as a
     contraction (note that this differs from Einstein convention; repeated
@@ -82,31 +102,35 @@ def blockwise(
     index.  To receive arrays concatenated along contracted dimensions instead
     pass ``concatenate=True``.
 
-    Inner product multiplying x by y, two 1-d vectors
+    Inner product multiplying a by b, two 1-d vectors
 
-    >>> def sequence_dot(x_blocks, y_blocks):
+    >>> def sequence_dot(a_blocks, b_blocks):
     ...     result = 0
-    ...     for x, y in zip(x_blocks, y_blocks):
-    ...         result += x.dot(y)
+    ...     for a, b in zip(a_blocks, b_blocks):
+    ...         result += a.dot(b)
     ...     return result
 
-    >>> z = blockwise(sequence_dot, '', x, 'i', y, 'i', dtype='f8')  # doctest: +SKIP
+    >>> z = blockwise(sequence_dot, '', a, 'i', b, 'i', dtype='f8')
+    >>> z.compute()
+    250
 
     Add new single-chunk dimensions with the ``new_axes=`` keyword, including
     the length of the new dimension.  New dimensions will always be in a single
     chunk.
 
-    >>> def f(x):
-    ...     return x[:, None] * np.ones((1, 5))
+    >>> def f(a):
+    ...     return a[:, None] * np.ones((1, 5))
 
-    >>> z = blockwise(f, 'az', x, 'a', new_axes={'z': 5}, dtype=x.dtype)  # doctest: +SKIP
+    >>> z = blockwise(f, 'az', a, 'a', new_axes={'z': 5}, dtype=a.dtype)
 
     New dimensions can also be multi-chunk by specifying a tuple of chunk
     sizes.  This has limited utility as is (because the chunks are all the
     same), but the resulting graph can be modified to achieve more useful
     results (see ``da.map_blocks``).
 
-    >>> z = blockwise(f, 'az', x, 'a', new_axes={'z': (5, 5)}, dtype=x.dtype)  # doctest: +SKIP
+    >>> z = blockwise(f, 'az', a, 'a', new_axes={'z': (5, 5)}, dtype=x.dtype)
+    >>> z.chunks
+    ((1, 1, 1), (5, 5))
 
     If the applied function changes the size of each chunk you can specify this
     with a ``adjust_chunks={...}`` dictionary holding a function for each index
@@ -116,11 +140,16 @@ def blockwise(
     ...     return np.concatenate([x, x])
 
     >>> y = blockwise(double, 'ij', x, 'ij',
-    ...               adjust_chunks={'i': lambda n: 2 * n}, dtype=x.dtype)  # doctest: +SKIP
+    ...               adjust_chunks={'i': lambda n: 2 * n}, dtype=x.dtype)
+    >>> y.chunks
+    ((2, 2), (2,))
 
     Include literals by indexing with None
 
-    >>> y = blockwise(add, 'ij', x, 'ij', 1234, None, dtype=x.dtype)  # doctest: +SKIP
+    >>> z = blockwise(operator.add, 'ij', x, 'ij', 1234, None, dtype=x.dtype)
+    >>> z.compute()
+    array([[1235, 1236],
+           [1237, 1238]])
     """
     out = name
     new_axes = new_axes or {}
diff --git a/dask/array/utils.py b/dask/array/utils.py
index 330fd4520..e5a2a2abd 100644
--- a/dask/array/utils.py
+++ b/dask/array/utils.py
@@ -206,7 +206,7 @@ def _check_dsk(dsk):
 
     dsk.validate()
     assert all(isinstance(k, (tuple, str)) for k in dsk.layers)
-    freqs = frequencies(concat(dsk.dicts.values()))
+    freqs = frequencies(concat(dsk.layers.values()))
     non_one = {k: v for k, v in freqs.items() if v != 1}
     assert not non_one, non_one
 
diff --git a/dask/blockwise.py b/dask/blockwise.py
index 24ecc9ccd..550d7fe4c 100644
--- a/dask/blockwise.py
+++ b/dask/blockwise.py
@@ -354,7 +354,7 @@ class Blockwise(Layer):
         global_dependencies = list(state["global_dependencies"])
 
         if state["annotations"]:
-            annotations.update(cls.expand_annotations(state["annotations"], raw.keys()))
+            cls.unpack_annotations(annotations, state["annotations"], raw.keys())
 
         raw = {stringify(k): stringify_collection_keys(v) for k, v in raw.items()}
         dsk.update(raw)
@@ -861,9 +861,9 @@ def optimize_blockwise(graph, keys=()):
 
 def _optimize_blockwise(full_graph, keys=()):
     keep = {k[0] if type(k) is tuple else k for k in keys}
-    layers = full_graph.dicts
+    layers = full_graph.layers
     dependents = reverse_dict(full_graph.dependencies)
-    roots = {k for k in full_graph.dicts if not dependents.get(k)}
+    roots = {k for k in full_graph.layers if not dependents.get(k)}
     stack = list(roots)
 
     out = {}
diff --git a/dask/dataframe/io/parquet/arrow.py b/dask/dataframe/io/parquet/arrow.py
index ca18b243b..d94d0d7bf 100644
--- a/dask/dataframe/io/parquet/arrow.py
+++ b/dask/dataframe/io/parquet/arrow.py
@@ -21,6 +21,8 @@ from .utils import (
     _normalize_index_columns,
     Engine,
     _analyze_paths,
+    _flatten_filters,
+    _row_groups_to_parts,
 )
 
 
@@ -352,15 +354,6 @@ def _get_pandas_metadata(schema):
         return {}
 
 
-def _flatten_filters(filters):
-    """Flatten DNF-formatted filters (list of tuples)"""
-    return (
-        set(flatten(tuple(flatten(filters, container=list)), container=tuple))
-        if filters
-        else []
-    )
-
-
 def _read_table_from_path(
     path,
     fs,
@@ -614,10 +607,7 @@ class ArrowDatasetEngine(Engine):
         # For pyarrow.dataset api, if we did not read directly from
         # fragments, we need to add the partitioned columns here.
         if partitions and isinstance(partitions, list):
-            if not isinstance(path_or_frag, pa_ds.ParquetFileFragment):
-                keys_dict = {k: v for (k, v) in partition_keys}
-            else:
-                keys_dict = {}
+            keys_dict = {k: v for (k, v) in partition_keys}
             for partition in partitions:
                 if partition.name not in arrow_table.schema.names:
                     # We read from file paths, so the partition
@@ -637,7 +627,7 @@ class ArrowDatasetEngine(Engine):
         # to categorigal manually for integer types.
         if partitions and isinstance(partitions, list):
             for partition in partitions:
-                if df[partition.name].dtype != pd.Categorical:
+                if df[partition.name].dtype.name != "category":
                     # We read directly from fragments, so the partition
                     # columns are already in our dataframe.  We just
                     # need to convert non-categorical types.
@@ -1267,8 +1257,12 @@ class ArrowDatasetEngine(Engine):
         # If the user has not specified `gather_statistics`,
         # we will only do so if there are specific columns in
         # need of statistics.
+        # NOTE: We cannot change `gather_statistics` from True
+        # to False (even if `stat_col_indices` is empty), in
+        # case a `chunksize` was specified, and the row-group
+        # statistics are needed for part aggregation.
         if gather_statistics is None:
-            gather_statistics = len(stat_col_indices.keys()) > 0
+            gather_statistics = bool(stat_col_indices)
         if split_row_groups is None:
             split_row_groups = False
 
@@ -1279,36 +1273,19 @@ class ArrowDatasetEngine(Engine):
         )
 
     @classmethod
-    def _process_metadata(
+    def _organize_row_groups(
         cls,
         metadata,
-        schema,
         split_row_groups,
         gather_statistics,
         stat_col_indices,
         filters,
-        categories,
-        partition_info,
-        data_path,
-        fs,
-        read_from_paths,
     ):
-        """Process row-groups and statistics.
+        """Organize row-groups by file.
 
-        This method is overridden in `ArrowLegacyEngine`.
+        This method is used by ArrowDatasetEngine._process_metadata
         """
 
-        partition_keys = partition_info["partition_keys"]
-        partition_obj = partition_info["partitions"]
-
-        # Check if we need to pass a fragment for each output partition
-        read_from_paths = read_from_paths or False
-        pass_frags = (
-            filters
-            and (not read_from_paths)
-            and _need_fragments(filters, partition_keys)
-        )
-
         # Get the number of row groups per file
         frag_map = {}
         single_rg_parts = int(split_row_groups) == 1
@@ -1357,7 +1334,6 @@ class ArrowDatasetEngine(Engine):
                         if name in statistics:
                             cmin = statistics[name]["min"]
                             cmax = statistics[name]["max"]
-                            cnull = 0  # Not yet available/needed
                             last = cmax_last.get(name, None)
                             if not filters:
                                 # Only think about bailing if we don't need
@@ -1384,11 +1360,10 @@ class ArrowDatasetEngine(Engine):
                                         "max": pd.Timestamp(cmax)
                                         if isinstance(cmax, datetime)
                                         else cmax,
-                                        "null_count": cnull,
                                     }
                                 )
                             else:
-                                cstats += [cmin, cmax, cnull]
+                                cstats += [cmin, cmax]
                             cmax_last[name] = cmax
                         else:
                             if single_rg_parts:
@@ -1400,143 +1375,116 @@ class ArrowDatasetEngine(Engine):
                         if not single_rg_parts:
                             file_row_group_column_stats[fpath].append(tuple(cstats))
 
-        # Construct `parts` and `stats`
-        parts = []
-        stats = []
-        if split_row_groups:
-            # Create parts from each file,
-            # limiting the number of row_groups in each piece
-            split_row_groups = int(split_row_groups)
-            for filename, row_groups in file_row_groups.items():
-                row_group_count = len(row_groups)
-                for i in range(0, row_group_count, split_row_groups):
-                    i_end = i + split_row_groups
-                    rg_list = row_groups[i:i_end]
-                    # Get full path (empty strings should be ignored)
-                    full_path = fs.sep.join(
-                        [p for p in [data_path, filename] if p != ""]
-                    )
-                    pkeys = partition_keys.get(full_path, None)
-                    if partition_obj and pkeys is None:
-                        continue  # This partition was filtered
-                    part = {
-                        "piece": (
-                            frag_map[(full_path, rg_list[0])]
-                            if pass_frags
-                            else full_path,
-                            rg_list,
-                            pkeys,
-                        ),
-                    }
-                    parts.append(part)
-                    if gather_statistics:
-                        stat = cls._aggregate_stats(
-                            filename,
-                            file_row_group_stats[filename][i:i_end],
-                            file_row_group_column_stats[filename][i:i_end],
-                            stat_col_indices,
-                        )
-                        stats.append(stat)
-        else:
-            for filename, row_groups in file_row_groups.items():
-                # Get full path (empty strings should be ignored)
-                full_path = fs.sep.join([p for p in [data_path, filename] if p != ""])
-                pkeys = partition_keys.get(full_path, None)
-                if partition_obj and pkeys is None:
-                    continue  # This partition was filtered
-                part = {
-                    "piece": (
-                        frag_map[(full_path, row_groups[0])]
-                        if pass_frags
-                        else full_path,
-                        row_groups,
-                        pkeys,
-                    ),
-                }
-                parts.append(part)
-                if gather_statistics:
-                    stat = cls._aggregate_stats(
-                        filename,
-                        file_row_group_stats[filename],
-                        file_row_group_column_stats[filename],
-                        stat_col_indices,
-                    )
-                    stats.append(stat)
+        return (
+            file_row_groups,
+            file_row_group_stats,
+            file_row_group_column_stats,
+            gather_statistics,
+            frag_map,
+        )
 
-        common_kwargs = {
-            "partitioning": partition_info["partitioning"],
-            "partitions": partition_obj,
-            "categories": categories,
-            "filters": filters,
-            "schema": schema,
-        }
+    @classmethod
+    def _make_part(
+        cls,
+        filename,
+        rg_list,
+        fs=None,
+        partition_keys=None,
+        partition_obj=None,
+        data_path=None,
+        frag_map=None,
+    ):
+        """Generate a partition-specific element of `parts`.
 
-        return parts, stats, common_kwargs
+        This method is used by both `ArrowDatasetEngine`
+        and `ArrowLegacyEngine`.
+        """
+
+        # Get full path (empty strings should be ignored)
+        full_path = fs.sep.join([p for p in [data_path, filename] if p != ""])
+
+        pkeys = partition_keys.get(full_path, None)
+        if partition_obj and pkeys is None:
+            return None  # This partition was filtered
+        return {
+            "piece": (
+                frag_map[(full_path, rg_list[0])] if frag_map else full_path,
+                rg_list,
+                pkeys,
+            ),
+        }
 
     @classmethod
-    def _aggregate_stats(
+    def _process_metadata(
         cls,
-        file_path,
-        file_row_group_stats,
-        file_row_group_column_stats,
+        metadata,
+        schema,
+        split_row_groups,
+        gather_statistics,
         stat_col_indices,
+        filters,
+        categories,
+        partition_info,
+        data_path,
+        fs,
+        read_from_paths,
     ):
-        """Utility to aggregate the statistics for N row-groups
-        into a single dictionary.
+        """Process row-groups and statistics.
 
-        Used by `_construct_parts`
+        This method is overridden in `ArrowLegacyEngine`.
         """
-        if len(file_row_group_stats) < 1:
-            # Empty statistics
-            return {}
-        elif len(file_row_group_column_stats) == 0:
-            assert len(file_row_group_stats) == 1
-            return file_row_group_stats[0]
-        else:
-            # Note: It would be better to avoid df_rgs and df_cols
-            #       construction altogether. It makes it fast to aggregate
-            #       the statistics for many row groups, but isn't
-            #       worthwhile for a small number of row groups.
-            if len(file_row_group_stats) > 1:
-                df_rgs = pd.DataFrame(file_row_group_stats)
-                s = {
-                    "file_path_0": file_path,
-                    "num-rows": df_rgs["num-rows"].sum(),
-                    "total_byte_size": df_rgs["total_byte_size"].sum(),
-                    "columns": [],
-                }
-            else:
-                s = {
-                    "file_path_0": file_path,
-                    "num-rows": file_row_group_stats[0]["num-rows"],
-                    "total_byte_size": file_row_group_stats[0]["total_byte_size"],
-                    "columns": [],
-                }
-
-            df_cols = None
-            if len(file_row_group_column_stats) > 1:
-                df_cols = pd.DataFrame(file_row_group_column_stats)
-            for ind, name in enumerate(stat_col_indices):
-                i = ind * 3
-                if df_cols is None:
-                    s["columns"].append(
-                        {
-                            "name": name,
-                            "min": file_row_group_column_stats[0][i],
-                            "max": file_row_group_column_stats[0][i + 1],
-                            "null_count": file_row_group_column_stats[0][i + 2],
-                        }
-                    )
-                else:
-                    s["columns"].append(
-                        {
-                            "name": name,
-                            "min": df_cols.iloc[:, i].min(),
-                            "max": df_cols.iloc[:, i + 1].max(),
-                            "null_count": df_cols.iloc[:, i + 2].sum(),
-                        }
-                    )
-            return s
+
+        # Organize row-groups by file
+        (
+            file_row_groups,
+            file_row_group_stats,
+            file_row_group_column_stats,
+            gather_statistics,
+            frag_map,
+        ) = cls._organize_row_groups(
+            metadata,
+            split_row_groups,
+            gather_statistics,
+            stat_col_indices,
+            filters,
+        )
+
+        # Check if we need to pass a fragment for each output partition
+        read_from_paths = read_from_paths or False
+        pass_frags = (
+            filters
+            and (not read_from_paths)
+            and _need_fragments(filters, partition_info.get("partition_keys", None))
+        )
+
+        # Convert organized row-groups to parts
+        parts, stats = _row_groups_to_parts(
+            gather_statistics,
+            split_row_groups,
+            file_row_groups,
+            file_row_group_stats,
+            file_row_group_column_stats,
+            stat_col_indices,
+            cls._make_part,
+            make_part_kwargs={
+                "fs": fs,
+                "partition_keys": partition_info.get("partition_keys", None),
+                "partition_obj": partition_info.get("partitions", None),
+                "data_path": data_path,
+                "frag_map": frag_map if pass_frags else None,
+            },
+        )
+
+        # Add common kwargs
+        common_kwargs = {
+            "partitioning": partition_info["partitioning"],
+            "partitions": partition_info["partitions"],
+            "categories": categories,
+            "filters": filters,
+            "schema": schema,
+        }
+
+        return parts, stats, common_kwargs
 
     @classmethod
     def _read_table(
@@ -1855,28 +1803,19 @@ class ArrowLegacyEngine(ArrowDatasetEngine):
             )
 
     @classmethod
-    def _process_metadata(
+    def _organize_row_groups(
         cls,
         metadata,
-        schema,
         split_row_groups,
         gather_statistics,
         stat_col_indices,
         filters,
-        categories,
-        partition_info,
-        data_path,
-        fs,
-        read_from_paths,
     ):
-        """Process row-groups and statistics.
+        """Organize row-groups by file.
 
-        This method is overrides the `ArrowDatasetEngine` implementation.
+        This method is used by ArrowLegacyEngine._process_metadata
         """
 
-        partition_keys = partition_info["partition_keys"]
-        partition_obj = partition_info["partitions"]
-
         # Get the number of row groups per file
         single_rg_parts = int(split_row_groups) == 1
         file_row_groups = defaultdict(list)
@@ -1885,6 +1824,10 @@ class ArrowLegacyEngine(ArrowDatasetEngine):
         cmax_last = {}
         for rg in range(metadata.num_row_groups):
             row_group = metadata.row_group(rg)
+
+            # NOTE: Here we assume that all column chunks are stored
+            # in the same file. This is not strictly required by the
+            # parquet spec.
             fpath = row_group.column(0).file_path
             if fpath is None:
                 raise ValueError(
@@ -1915,7 +1858,6 @@ class ArrowLegacyEngine(ArrowDatasetEngine):
                     if column.statistics:
                         cmin = column.statistics.min
                         cmax = column.statistics.max
-                        cnull = column.statistics.null_count
                         last = cmax_last.get(name, None)
                         if not filters:
                             # Only think about bailing if we don't need
@@ -1939,11 +1881,10 @@ class ArrowLegacyEngine(ArrowDatasetEngine):
                                     "name": name,
                                     "min": cmin if not to_ts else pd.Timestamp(cmin),
                                     "max": cmax if not to_ts else pd.Timestamp(cmax),
-                                    "null_count": cnull,
                                 }
                             )
                         else:
-                            cstats += [cmin, cmax, cnull]
+                            cstats += [cmin, cmax]
                         cmax_last[name] = cmax
                     else:
 
@@ -1964,55 +1905,67 @@ class ArrowLegacyEngine(ArrowDatasetEngine):
                     if not single_rg_parts:
                         file_row_group_column_stats[fpath].append(tuple(cstats))
 
-        # Construct `parts` and `stats`
-        parts = []
-        stats = []
-        if split_row_groups:
-            # Create parts from each file,
-            # limiting the number of row_groups in each piece
-            split_row_groups = int(split_row_groups)
-            for filename, row_groups in file_row_groups.items():
-                row_group_count = len(row_groups)
-                for i in range(0, row_group_count, split_row_groups):
-                    i_end = i + split_row_groups
-                    rg_list = row_groups[i:i_end]
-                    # Get full path (empty strings should be ignored)
-                    full_path = fs.sep.join(
-                        [p for p in [data_path, filename] if p != ""]
-                    )
-                    pkeys = partition_keys.get(full_path, None)
-                    if partition_obj and pkeys is None:
-                        continue  # This partition was filtered
-                    part = {"piece": (full_path, rg_list, pkeys)}
-                    parts.append(part)
-                    if gather_statistics:
-                        stat = cls._aggregate_stats(
-                            filename,
-                            file_row_group_stats[filename][i:i_end],
-                            file_row_group_column_stats[filename][i:i_end],
-                            stat_col_indices,
-                        )
-                        stats.append(stat)
-        else:
-            for filename, row_groups in file_row_groups.items():
-                # Get full path (empty strings should be ignored)
-                full_path = fs.sep.join([p for p in [data_path, filename] if p != ""])
-                pkeys = partition_keys.get(full_path, None)
-                if partition_obj and pkeys is None:
-                    continue  # This partition was filtered
-                part = {"piece": (full_path, None, pkeys)}
-                parts.append(part)
-                if gather_statistics:
-                    stat = cls._aggregate_stats(
-                        filename,
-                        file_row_group_stats[filename],
-                        file_row_group_column_stats[filename],
-                        stat_col_indices,
-                    )
-                    stats.append(stat)
+        return (
+            file_row_groups,
+            file_row_group_stats,
+            file_row_group_column_stats,
+            gather_statistics,
+        )
+
+    @classmethod
+    def _process_metadata(
+        cls,
+        metadata,
+        schema,
+        split_row_groups,
+        gather_statistics,
+        stat_col_indices,
+        filters,
+        categories,
+        partition_info,
+        data_path,
+        fs,
+        read_from_paths,
+    ):
+        """Process row-groups and statistics.
+
+        This method is overrides the `ArrowDatasetEngine` implementation.
+        """
+
+        # Organize row-groups by file
+        (
+            file_row_groups,
+            file_row_group_stats,
+            file_row_group_column_stats,
+            gather_statistics,
+        ) = cls._organize_row_groups(
+            metadata,
+            split_row_groups,
+            gather_statistics,
+            stat_col_indices,
+            filters,
+        )
+
+        # Convert organized row-groups to parts
+        parts, stats = _row_groups_to_parts(
+            gather_statistics,
+            split_row_groups,
+            file_row_groups,
+            file_row_group_stats,
+            file_row_group_column_stats,
+            stat_col_indices,
+            cls._make_part,
+            make_part_kwargs={
+                "fs": fs,
+                "partition_keys": partition_info.get("partition_keys", None),
+                "partition_obj": partition_info.get("partitions", None),
+                "data_path": data_path,
+            },
+        )
 
+        # Add common kwargs
         common_kwargs = {
-            "partitions": partition_obj,
+            "partitions": partition_info["partitions"],
             "categories": categories,
             "filters": filters,
         }
diff --git a/dask/dataframe/io/parquet/core.py b/dask/dataframe/io/parquet/core.py
index 852de4f09..cf13e3f4c 100644
--- a/dask/dataframe/io/parquet/core.py
+++ b/dask/dataframe/io/parquet/core.py
@@ -62,6 +62,7 @@ class ParquetSubgraph(Layer):
         self.columns = columns
         self.index = index
         self.parts = parts
+        self.kwargs = kwargs
         self.part_ids = list(range(len(parts))) if part_ids is None else part_ids
 
         # `kwargs` = user-defined kwargs to be passed for all parts
@@ -97,12 +98,10 @@ class ParquetSubgraph(Layer):
             self.fs,
             self.engine.read_partition,
             self.meta,
-            [p["piece"] for p in part],
+            [(p["piece"], p.get("kwargs", {})) for p in part],
             self.columns,
             self.index,
-            toolz.merge(
-                self.common_kwargs, part[0].get("kwargs", {}), self.kwargs or {}
-            ),
+            toolz.merge(self.common_kwargs, self.kwargs or {}),
         )
 
     def __len__(self):
@@ -175,15 +174,18 @@ def read_parquet(
         filtering is only performed at the partition level, i.e., to prevent the
         loading of some row-groups and/or files.
 
-        Predicates can be expressed in disjunctive normal form (DNF). This means
-        that the innermost tuple describes a single column predicate. These
-        inner predicates are combined with an AND conjunction into a larger
-        predicate. The outer-most list then combines all of the combined
-        filters with an OR disjunction.
+        For the "pyarrow" engines, predicates can be expressed in disjunctive
+        normal form (DNF). This means that the innermost tuple describes a single
+        column predicate. These inner predicates are combined with an AND
+        conjunction into a larger predicate. The outer-most list then combines all
+        of the combined filters with an OR disjunction.
 
         Predicates can also be expressed as a List[Tuple]. These are evaluated
         as an AND conjunction. To express OR in predictates, one must use the
-        (preferred) List[List[Tuple]] notation.
+        (preferred for "pyarrow") List[List[Tuple]] notation.
+
+        Note that the "fastparquet" engine does not currently support DNF for
+        the filtering of partitioned columns (List[Tuple] is required).
     index : string, list, False or None (default)
         Field name(s) to use as the output frame index. By default will be
         inferred from the pandas parquet file metadata (if present). Use False
@@ -378,10 +380,16 @@ def read_parquet_part(fs, func, meta, part, columns, index, kwargs):
     This function is used by `read_parquet`."""
 
     if isinstance(part, list):
-        dfs = [func(fs, rg, columns.copy(), index, **kwargs) for rg in part]
+        dfs = [
+            func(fs, rg, columns.copy(), index, **toolz.merge(kwargs, kw))
+            for (rg, kw) in part
+        ]
         df = concat(dfs, axis=0)
     else:
-        df = func(fs, part, columns, index, **kwargs)
+        # NOTE: `kwargs` are the same for all parts, while `part_kwargs` may
+        #       be different for each part.
+        rg, part_kwargs = part
+        df = func(fs, rg, columns, index, **toolz.merge(kwargs, part_kwargs))
 
     if meta.columns.name:
         df.columns.name = meta.columns.name
@@ -935,6 +943,8 @@ def apply_filters(parts, statistics, filters):
         as an AND conjunction. To express OR in predictates, one must use the
         (preferred) List[List[Tuple]] notation.
 
+        Note that the "fastparquet" engine does not currently support DNF for
+        the filtering of partitioned columns (List[Tuple] is required).
     Returns
     -------
     parts, statistics: the same as the input, but possibly a subset
@@ -1138,8 +1148,6 @@ def aggregate_row_groups(parts, stats, chunksize):
             for col, col_add in zip(next_stat["columns"], stat["columns"]):
                 if col["name"] != col_add["name"]:
                     raise ValueError("Columns are different!!")
-                if "null_count" in col:
-                    col["null_count"] += col_add["null_count"]
                 if "min" in col:
                     col["min"] = min(col["min"], col_add["min"])
                 if "max" in col:
diff --git a/dask/dataframe/io/parquet/fastparquet.py b/dask/dataframe/io/parquet/fastparquet.py
index 94039267b..594d945ce 100644
--- a/dask/dataframe/io/parquet/fastparquet.py
+++ b/dask/dataframe/io/parquet/fastparquet.py
@@ -1,8 +1,10 @@
 from distutils.version import LooseVersion
+from collections import defaultdict
 
 from collections import OrderedDict
 import copy
 import json
+import pickle
 import warnings
 
 import tlz as toolz
@@ -19,9 +21,17 @@ try:
 except ImportError:
     pass
 
-from .utils import _parse_pandas_metadata, _normalize_index_columns, _analyze_paths
+from .utils import (
+    _parse_pandas_metadata,
+    _normalize_index_columns,
+    _analyze_paths,
+    _flatten_filters,
+    _row_groups_to_parts,
+)
 from ..utils import _meta_from_dtypes
 from ...utils import UNKNOWN_CATEGORIES
+from ...methods import concat
+
 
 #########################
 # Fastparquet interface #
@@ -103,17 +113,8 @@ def _determine_pf_parts(fs, paths, gather_statistics, **kwargs):
     because this also means we should avoid scanning every file in the
     dataset.  If _metadata is available, set `gather_statistics=True`
     (if `gather_statistics=None`).
-
-    The `fast_metadata` output specifies that ParquetFile metadata parsing
-    is fast enough for each worker to perform during `read_partition`. The
-    value will be set to True if: (1) The path is a directory containing
-    _metadta, (2) the path is a list of files containing _metadata, (3)
-    there is only one file to read, or (4) `gather_statistics` is False.
-    In other cases, the ParquetFile object will need to be stored in the
-    task graph, because metadata parsing is too expensive.
     """
     parts = []
-    fast_metadata = True
     if len(paths) > 1:
         base, fns = _analyze_paths(paths, fs)
         if gather_statistics is not False:
@@ -121,7 +122,6 @@ def _determine_pf_parts(fs, paths, gather_statistics, **kwargs):
             # and filtering
             if "_metadata" not in fns:
                 paths_use = paths
-                fast_metadata = False
             else:
                 paths_use = base + fs.sep + "_metadata"
             pf = ParquetFile(
@@ -134,7 +134,7 @@ def _determine_pf_parts(fs, paths, gather_statistics, **kwargs):
                     base + fs.sep + "_metadata",
                     open_with=fs.open,
                     sep=fs.sep,
-                    **kwargs.get("file", {})
+                    **kwargs.get("file", {}),
                 )
             else:
                 # Rely on metadata for 0th file.
@@ -154,7 +154,7 @@ def _determine_pf_parts(fs, paths, gather_statistics, **kwargs):
                 base + fs.sep + "_metadata",
                 open_with=fs.open,
                 sep=fs.sep,
-                **kwargs.get("file", {})
+                **kwargs.get("file", {}),
             )
             if gather_statistics is None:
                 gather_statistics = True
@@ -162,7 +162,6 @@ def _determine_pf_parts(fs, paths, gather_statistics, **kwargs):
         elif gather_statistics is not False:
             # Scan every file
             pf = ParquetFile(paths, open_with=fs.open, **kwargs.get("file", {}))
-            fast_metadata = False
         else:
             # Use _common_metadata file if it is available.
             # Otherwise, just use 0th file
@@ -170,14 +169,15 @@ def _determine_pf_parts(fs, paths, gather_statistics, **kwargs):
                 pf = ParquetFile(
                     base + fs.sep + "_common_metadata",
                     open_with=fs.open,
-                    **kwargs.get("file", {})
+                    **kwargs.get("file", {}),
                 )
+                fns.remove("_common_metadata")
             else:
                 pf = ParquetFile(paths[0], open_with=fs.open, **kwargs.get("file", {}))
             scheme = get_file_scheme(fns)
             pf.file_scheme = scheme
             pf.cats = paths_to_cats(fns, scheme)
-            parts = paths.copy()
+            parts = [fs.sep.join([base, fn]) for fn in fns]
     else:
         # There is only one file to read
         base = None
@@ -185,29 +185,12 @@ def _determine_pf_parts(fs, paths, gather_statistics, **kwargs):
             paths[0], open_with=fs.open, sep=fs.sep, **kwargs.get("file", {})
         )
 
-    return parts, pf, gather_statistics, fast_metadata, base
+    return parts, pf, gather_statistics, base
 
 
 class FastParquetEngine(Engine):
     @classmethod
-    def read_metadata(
-        cls,
-        fs,
-        paths,
-        categories=None,
-        index=None,
-        gather_statistics=None,
-        filters=None,
-        **kwargs
-    ):
-        # Define the parquet-file (pf) object to use for metadata,
-        # Also, initialize `parts`.  If `parts` is populated here,
-        # then each part will correspond to a file.  Otherwise, each part will
-        # correspond to a row group (populated below).
-        parts, pf, gather_statistics, fast_metadata, base_path = _determine_pf_parts(
-            fs, paths, gather_statistics, **kwargs
-        )
-
+    def _generate_dd_meta(cls, pf, index, categories):
         columns = None
         if pf.fmd.key_value_metadata:
             pandas_md = [
@@ -271,6 +254,8 @@ class FastParquetEngine(Engine):
 
         index_cols = index or ()
         meta = _meta_from_dtypes(all_columns, dtypes, index_cols, column_index_names)
+        if isinstance(index_cols, str):
+            index_cols = [index_cols]
 
         # fastparquet doesn't handle multiindex
         if len(index_names) > 1:
@@ -288,154 +273,441 @@ class FastParquetEngine(Engine):
                 meta[catcol] = meta[catcol].cat.set_categories(pf.cats[catcol])
             elif meta.index.name == catcol:
                 meta.index = meta.index.set_categories(pf.cats[catcol])
-        if gather_statistics and pf.row_groups:
-            stats = []
-            if filters is None:
-                filters = []
-
-            skip_cols = set()  # Columns with min/max = None detected
-            # make statistics conform in layout
-            for (i, row_group) in enumerate(pf.row_groups):
-                s = {"num-rows": row_group.num_rows, "columns": []}
-                for i_col, col in enumerate(pf.columns):
-                    if col not in skip_cols:
-                        d = {"name": col}
-                        cs_min = None
-                        cs_max = None
-                        if pf.statistics["min"][col][0] is not None:
-                            cs_min = pf.statistics["min"][col][i]
-                            cs_max = pf.statistics["max"][col][i]
-                        elif (
-                            dtypes[col] == "object"
-                            and row_group.columns[i_col].meta_data.statistics
-                        ):
-                            cs_min = row_group.columns[
-                                i_col
-                            ].meta_data.statistics.min_value
-                            cs_max = row_group.columns[
-                                i_col
-                            ].meta_data.statistics.max_value
-                            if isinstance(cs_min, (bytes, bytearray)):
-                                cs_min = cs_min.decode("utf-8")
-                                cs_max = cs_max.decode("utf-8")
-                        if None in [cs_min, cs_max] and i == 0:
-                            skip_cols.add(col)
-                            continue
-                        if isinstance(cs_min, np.datetime64):
-                            tz = getattr(dtypes[col], "tz", None)
-                            cs_min = pd.Timestamp(cs_min, tz=tz)
-                            cs_max = pd.Timestamp(cs_max, tz=tz)
-                        d.update(
-                            {
-                                "min": cs_min,
-                                "max": cs_max,
-                                "null_count": pf.statistics["null_count"][col][i],
-                            }
-                        )
-                        s["columns"].append(d)
-                # Need this to filter out partitioned-on categorical columns
-                s["filter"] = fastparquet.api.filter_out_cats(row_group, filters)
-                s["total_byte_size"] = row_group.total_byte_size
-                s["file_path_0"] = row_group.columns[0].file_path  # 0th column only
-                stats.append(s)
 
-        else:
-            stats = None
-
-        pf._dtypes = lambda *args: pf.dtypes  # ugly patch, could be fixed
-        pf.fmd.row_groups = None
-
-        # Constructing "piece" and "pf" for each dask partition. We will
-        # need to consider the following output scenarios:
-        #
-        #  1) Each "piece" is a file path, and "pf" is `None`
-        #      - `gather_statistics==False` and no "_metadata" available
-        #  2) Each "piece" is a row-group index, and "pf" is ParquetFile object
-        #      - We have parquet partitions and no "_metadata" available
-        #  3) Each "piece" is a row-group index, and "pf" is a `tuple`
-        #      - The value of the 0th tuple element depends on the following:
-        #      A) Dataset is partitioned and "_metadata" exists
-        #          - 0th tuple element will be the path to "_metadata"
-        #      B) Dataset is not partitioned
-        #          - 0th tuple element will be the path to the data
-        #      C) Dataset is partitioned and "_metadata" does not exist
-        #          - 0th tuple element will be the original `paths` argument
-        #            (We will let the workers use `_determine_pf_parts`)
-
-        # Create `parts`
-        # This is a list of row-group-descriptor dicts, or file-paths
-        # if we have a list of files and gather_statistics=False
-        base_path = (base_path or "") + fs.sep
-        if parts:
-            # Case (1)
-            # `parts` is just a list of path names.
-            pqpartitions = None
-            pf_deps = None
-            partsin = parts
-        else:
-            # Populate `partsin` with dataset row-groups
-            partsin = pf.row_groups
-            pqpartitions = pf.info.get("partitions", None)
-            if pqpartitions and not fast_metadata:
-                # Case (2)
-                # We have parquet partitions, and do not have
-                # a "_metadata" file for the worker to read.
-                # Therefore, we need to pass the pf object in
-                # the task graph
-                pf_deps = pf
-            else:
-                # Case (3)
-                # We don't need to pass a pf object in the task graph.
-                # Instead, we can try to pass the path for each part.
-                pf_deps = "tuple"
-
-        parts = []
-        i_path = 0
-        path_last = None
-
-        # Loop over DataFrame partitions.
-        # Each `part` will be a row-group or path ()
-        for i, part in enumerate(partsin):
-            if pqpartitions and fast_metadata:
-                # Case (3A)
-                # We can pass a "_metadata" path
-                file_path = base_path + "_metadata"
-                i_path = i
-            elif (
-                pf_deps
-                and isinstance(part.columns[0].file_path, str)
-                and not pqpartitions
+        return meta, dtypes, index_cols, categories_dict, categories, index
+
+    @classmethod
+    def _update_metadata_options(
+        cls,
+        pf,
+        parts,
+        gather_statistics,
+        split_row_groups,
+        index_cols,
+        filters,
+    ):
+        # Cannot gather_statistics if our `parts` is alreadey a list
+        # of paths, or if we are building a multiindex (for now).
+        # We also don't "need" to gather statistics if we don't
+        # want to apply any filters or calculate divisions.
+        if (
+            isinstance(parts, list) and len(parts) and isinstance(parts[0], str)
+        ) or len(index_cols) > 1:
+            gather_statistics = False
+        elif filters is None and len(index_cols) == 0:
+            gather_statistics = False
+
+        # Make sure gather_statistics allows filtering
+        # (if filters are desired)
+        if filters:
+            # Filters may require us to gather statistics
+            if gather_statistics is False and pf.info.get("partitions", None):
+                warnings.warn(
+                    "Filtering with gather_statistics=False. "
+                    "Only partition columns will be filtered correctly."
+                )
+            elif gather_statistics is False:
+                raise ValueError("Cannot apply filters with gather_statistics=False")
+            elif not gather_statistics:
+                gather_statistics = True
+
+        # Determine which columns need statistics.
+        flat_filters = _flatten_filters(filters)
+        stat_col_indices = {}
+        for i, name in enumerate(pf.columns):
+            if name in index_cols or name in flat_filters:
+                stat_col_indices[name] = i
+
+        # If the user has not specified `gather_statistics`,
+        # we will only do so if there are specific columns in
+        # need of statistics.
+        # NOTE: We cannot change `gather_statistics` from True
+        # to False (even if `stat_col_indices` is empty), in
+        # case a `chunksize` was specified, and the row-group
+        # statistics are needed for part aggregation.
+        if gather_statistics is None:
+            gather_statistics = bool(stat_col_indices)
+        if split_row_groups is None:
+            split_row_groups = False
+
+        return (
+            gather_statistics,
+            split_row_groups,
+            stat_col_indices,
+        )
+
+    @classmethod
+    def _organize_row_groups(
+        cls,
+        pf,
+        split_row_groups,
+        gather_statistics,
+        stat_col_indices,
+        filters,
+        dtypes,
+        base_path,
+        paths,
+    ):
+        """Organize row-groups by file."""
+
+        # Get partitioning metadata
+        pqpartitions = pf.info.get("partitions", None)
+
+        # Store types specified in pandas metadata
+        pandas_type = {}
+        if pf.row_groups and pf.pandas_metadata:
+            for c in pf.pandas_metadata.get("columns", []):
+                if "field_name" in c:
+                    pandas_type[c["field_name"]] = c.get("pandas_type", None)
+
+        # Get the number of row groups per file
+        single_rg_parts = int(split_row_groups) == 1
+        file_row_groups = defaultdict(list)
+        file_row_group_stats = defaultdict(list)
+        file_row_group_column_stats = defaultdict(list)
+        cmax_last = {}
+        for rg, row_group in enumerate(pf.row_groups):
+
+            # We can filter partition columns here without dealing
+            # with statistics
+            if (
+                pqpartitions
+                and filters
+                and fastparquet.api.filter_out_cats(row_group, filters)
             ):
-                # Case (3B)
-                # We can pass a specific file/part path
-                path_curr = part.columns[0].file_path
-                if path_last and path_curr == path_last:
-                    i_path += 1
+                continue
+
+            # NOTE: Here we assume that all column chunks are stored
+            # in the same file. This is not strictly required by the
+            # parquet spec.
+            fpath = row_group.columns[0].file_path
+            if fpath is None:
+                if paths and pf.fn in paths:
+                    # There doesn't need to be a file_path if the
+                    # row group is in the same file as the metadata.
+                    # Assume this is a single-file dataset.
+                    fpath = pf.fn
+                    base_path = base_path or ""
                 else:
-                    i_path = 0
-                    path_last = path_curr
-                file_path = base_path + path_curr
+                    raise ValueError(
+                        "Global metadata structure is missing a file_path string. "
+                        "If the dataset includes a _metadata file, that file may "
+                        "have one or more missing file_path fields."
+                    )
+
+            # Append a tuple to file_row_groups. This tuple will
+            # be structured as: `(<local-row-group-id>, <global-row-group-id>)`
+            if file_row_groups[fpath]:
+                file_row_groups[fpath].append((file_row_groups[fpath][-1][0] + 1, rg))
             else:
-                # Case (3C)
-                # We cannot pass a specific file/part path
-                file_path = paths
-                i_path = i
-
-            # Strip down pf object
-            if pf_deps and pf_deps != "tuple":
-                # Case (2)
-                for col in part.columns:
-                    col.meta_data.statistics = None
-                    col.meta_data.encoding_stats = None
-
-            # Final definition of "piece" and "pf" for this output partition
-            piece = i_path if pf_deps else part
-            pf_piece = (file_path, gather_statistics) if pf_deps == "tuple" else pf_deps
-            part_item = {
-                "piece": piece,
-                "kwargs": {"pf": pf_piece},
-            }
-            parts.append(part_item)
+                file_row_groups[fpath].append((0, rg))
+
+            if gather_statistics:
+                if single_rg_parts:
+                    s = {
+                        "file_path_0": fpath,
+                        "num-rows": row_group.num_rows,
+                        "total_byte_size": row_group.total_byte_size,
+                        "columns": [],
+                    }
+                else:
+                    s = {
+                        "num-rows": row_group.num_rows,
+                        "total_byte_size": row_group.total_byte_size,
+                    }
+                cstats = []
+                for name, i in stat_col_indices.items():
+                    column = row_group.columns[i]
+                    if column.meta_data.statistics:
+                        cmin = None
+                        cmax = None
+                        # TODO: Avoid use of `pf.statistics`
+                        if pf.statistics["min"][name][0] is not None:
+                            cmin = pf.statistics["min"][name][rg]
+                            cmax = pf.statistics["max"][name][rg]
+                        elif dtypes[name] == "object":
+                            cmin = column.meta_data.statistics.min_value
+                            cmax = column.meta_data.statistics.max_value
+                            # Older versions may not have cmin/cmax_value
+                            if cmin is None:
+                                cmin = column.meta_data.statistics.min
+                            if cmax is None:
+                                cmax = column.meta_data.statistics.max
+                            # Decode bytes as long as "bytes" is not the
+                            # expected `pandas_type` for this column
+                            if (
+                                isinstance(cmin, (bytes, bytearray))
+                                and pandas_type.get(name, None) != "bytes"
+                            ):
+                                cmin = cmin.decode("utf-8")
+                                cmax = cmax.decode("utf-8")
+                        if isinstance(cmin, np.datetime64):
+                            tz = getattr(dtypes[name], "tz", None)
+                            cmin = pd.Timestamp(cmin, tz=tz)
+                            cmax = pd.Timestamp(cmax, tz=tz)
+                        last = cmax_last.get(name, None)
+
+                        if not filters:
+                            # Only think about bailing if we don't need
+                            # stats for filtering
+                            if cmin is None or (last and cmin < last):
+                                # We are collecting statistics for divisions
+                                # only (no filters) - Column isn't sorted, or
+                                # we have an all-null partition, so lets bail.
+                                #
+                                # Note: This assumes ascending order.
+                                #
+                                gather_statistics = False
+                                file_row_group_stats = {}
+                                file_row_group_column_stats = {}
+                                break
+
+                        if single_rg_parts:
+                            s["columns"].append(
+                                {
+                                    "name": name,
+                                    "min": cmin,
+                                    "max": cmax,
+                                }
+                            )
+                        else:
+                            cstats += [cmin, cmax]
+                        cmax_last[name] = cmax
+                    else:
+                        if not filters and column.meta_data.num_values > 0:
+                            # We are collecting statistics for divisions
+                            # only (no filters) - Lets bail.
+                            gather_statistics = False
+                            file_row_group_stats = {}
+                            file_row_group_column_stats = {}
+                            break
+
+                        if single_rg_parts:
+                            s["columns"].append({"name": name})
+                        else:
+                            cstats += [None, None, None]
+                if gather_statistics:
+                    file_row_group_stats[fpath].append(s)
+                    if not single_rg_parts:
+                        file_row_group_column_stats[fpath].append(tuple(cstats))
+
+        return (
+            file_row_groups,
+            file_row_group_stats,
+            file_row_group_column_stats,
+            gather_statistics,
+            base_path,
+        )
+
+    @classmethod
+    def _get_thrift_row_groups(
+        cls,
+        pf,
+        filename,
+        row_groups,
+    ):
+        """Turn a set of row-groups into bytes-serialized form
+        using thrift via pickle.
+        """
+
+        real_row_groups = []
+        for rg, rg_global in row_groups:
+            row_group = pf.row_groups[rg_global]
+            row_group.statistics = None
+            row_group.helper = None
+            for c, col in enumerate(row_group.columns):
+                if c:
+                    col.file_path = None
+                col.meta_data.key_value_metadata = None
+                # NOTE: Fastparquet may need the null count in the
+                # statistics, so we cannot just set statistics
+                # to none.  Set attributes separately:
+                if col.meta_data.statistics:
+                    col.meta_data.statistics.distinct_count = None
+                    col.meta_data.statistics.max = None
+                    col.meta_data.statistics.min = None
+                    col.meta_data.statistics.max_value = None
+                    col.meta_data.statistics.min_value = None
+                col.meta_data.encodings = None
+                col.meta_data.total_uncompressed_size = None
+                col.meta_data.encoding_stats = None
+            real_row_groups.append(row_group)
+        return pickle.dumps(real_row_groups)
+
+    @classmethod
+    def _make_part(
+        cls,
+        filename,
+        rg_list,
+        fs=None,
+        pf=None,
+        base_path=None,
+        partitions=None,
+    ):
+        """Generate a partition-specific element of `parts`."""
+
+        if partitions:
+            real_row_groups = cls._get_thrift_row_groups(
+                pf,
+                filename,
+                rg_list,
+            )
+            part = {"piece": (real_row_groups,)}
+        else:
+            # Get full path (empty strings should be ignored)
+            full_path = fs.sep.join([p for p in [base_path, filename] if p != ""])
+            row_groups = [rg[0] for rg in rg_list]  # Don't need global IDs
+            part = {"piece": (full_path, row_groups)}
+
+        return part
+
+    @classmethod
+    def _process_metadata(
+        cls,
+        pf,
+        dtypes,
+        split_row_groups,
+        gather_statistics,
+        stat_col_indices,
+        filters,
+        categories,
+        base_path,
+        paths,
+        fs,
+    ):
+
+        # Organize row-groups by file
+        (
+            file_row_groups,
+            file_row_group_stats,
+            file_row_group_column_stats,
+            gather_statistics,
+            base_path,
+        ) = cls._organize_row_groups(
+            pf,
+            split_row_groups,
+            gather_statistics,
+            stat_col_indices,
+            filters,
+            dtypes,
+            base_path,
+            paths,
+        )
+
+        # Convert organized row-groups to parts
+        parts, stats = _row_groups_to_parts(
+            gather_statistics,
+            split_row_groups,
+            file_row_groups,
+            file_row_group_stats,
+            file_row_group_column_stats,
+            stat_col_indices,
+            cls._make_part,
+            make_part_kwargs={
+                "fs": fs,
+                "pf": pf,
+                "base_path": base_path,
+                "partitions": pf.info.get("partitions", None),
+            },
+        )
+
+        return parts, stats
+
+    @classmethod
+    def _construct_parts(
+        cls,
+        fs,
+        pf,
+        paths,
+        parts,
+        dtypes,
+        base_path,
+        filters,
+        index_cols,
+        categories,
+        split_row_groups,
+        gather_statistics,
+    ):
+
+        # Check if `parts` is just a list of paths
+        # (not splitting by row-group or collecting statistics)
+        if isinstance(parts, list) and len(parts) and isinstance(parts[0], str):
+            stats = []
+            return [{"piece": (full_path, None)} for full_path in parts], stats
+
+        # Update `gather_statistics` and `split_row_groups`
+        # before constructing `parts`
+        (
+            gather_statistics,
+            split_row_groups,
+            stat_col_indices,
+        ) = cls._update_metadata_options(
+            pf,
+            parts,
+            gather_statistics,
+            split_row_groups,
+            index_cols,
+            filters,
+        )
+
+        # Process row-groups and return `(parts, stats)`
+        return cls._process_metadata(
+            pf,
+            dtypes,
+            split_row_groups,
+            gather_statistics,
+            stat_col_indices,
+            filters,
+            categories,
+            base_path,
+            paths,
+            fs,
+        )
+
+    @classmethod
+    def read_metadata(
+        cls,
+        fs,
+        paths,
+        categories=None,
+        index=None,
+        gather_statistics=None,
+        filters=None,
+        split_row_groups=True,
+        **kwargs,
+    ):
+        # Define the parquet-file (pf) object to use for metadata,
+        # Also, initialize `parts`.  If `parts` is populated here,
+        # then each part will correspond to a file.  Otherwise, each part will
+        # correspond to a row group (populated below).
+        parts, pf, gather_statistics, base_path = _determine_pf_parts(
+            fs, paths, gather_statistics, **kwargs
+        )
+
+        # Process metadata to define `meta` and `index_cols`
+        (
+            meta,
+            dtypes,
+            index_cols,
+            categories_dict,
+            categories,
+            index,
+        ) = cls._generate_dd_meta(pf, index, categories)
+
+        # Break `pf` into a list of `parts`
+        parts, stats = cls._construct_parts(
+            fs,
+            pf,
+            paths,
+            parts,
+            dtypes,
+            base_path,
+            filters,
+            index_cols,
+            categories,
+            split_row_groups,
+            gather_statistics,
+        )
 
         # Cannot allow `None` in columns if the user has specified index=False
         if index is False and None in meta.columns:
@@ -447,12 +719,19 @@ class FastParquetEngine(Engine):
         if len(parts):
             parts[0]["common_kwargs"] = {"categories": categories_dict or categories}
 
+        if len(parts) and len(parts[0]["piece"]) == 1:
+
+            # Strip all partition-dependent or unnecessary
+            # data from the `ParquetFile` object
+            pf.row_groups = None
+            pf.fmd.row_groups = None
+            pf._statistics = None
+            parts[0]["common_kwargs"]["parquet_file"] = pf
+
         return (meta, stats, parts, index)
 
     @classmethod
-    def read_partition(
-        cls, fs, piece, columns, index, categories=(), pf=None, **kwargs
-    ):
+    def read_partition(cls, fs, piece, columns, index, categories=(), **kwargs):
 
         null_index_name = False
         if isinstance(index, list):
@@ -468,44 +747,62 @@ class FastParquetEngine(Engine):
                 null_index_name = True
             columns += index
 
-        if pf is None:
-            base, fns = _analyze_paths([piece], fs)
-            scheme = get_file_scheme(fns)
-            pf = ParquetFile(piece, open_with=fs.open)
-            relpath = piece.replace(base, "").lstrip("/")
-            for rg in pf.row_groups:
-                for ch in rg.columns:
-                    ch.file_path = relpath
-            pf.file_scheme = scheme
-            pf.cats = paths_to_cats(fns, scheme)
-            pf.fn = base
-            if null_index_name and "__index_level_0__" in pf.columns:
-                # See "Handling a None-labeled index" comment above
-                index = ["__index_level_0__"]
-                columns += index
-            return pf.to_pandas(columns, categories, index=index)
-        else:
-            if isinstance(pf, tuple):
-                if isinstance(pf[0], list):
-                    pf = _determine_pf_parts(fs, pf[0], pf[1], **kwargs)[1]
-                else:
-                    pf = ParquetFile(
-                        pf[0], open_with=fs.open, sep=fs.sep, **kwargs.get("file", {})
-                    )
-                pf._dtypes = lambda *args: pf.dtypes  # ugly patch, could be fixed
-                pf.fmd.row_groups = None
-            rg_piece = pf.row_groups[piece]
+        # Use global `parquet_file` object.  Need to reattach
+        # the desired row_group
+        parquet_file = kwargs.pop("parquet_file", None)
+
+        if isinstance(piece, tuple):
+            if isinstance(piece[0], str):
+                # We have a path to read from
+                assert parquet_file is None
+                parquet_file = ParquetFile(
+                    piece[0], open_with=fs.open, sep=fs.sep, **kwargs.get("file", {})
+                )
+                rg_indices = piece[1] or list(range(len(parquet_file.row_groups)))
+
+                # `piece[1]` will contain row-group indices
+                row_groups = [parquet_file.row_groups[rg] for rg in rg_indices]
+            elif parquet_file:
+                # `piece[1]` will contain actual row-group objects,
+                # but they may be pickled
+                row_groups = piece[0]
+                if isinstance(row_groups, bytes):
+                    row_groups = pickle.loads(row_groups)
+                parquet_file.fmd.row_groups = row_groups
+                # NOTE: May lose cats after `_set_attrs` call
+                save_cats = parquet_file.cats
+                parquet_file._set_attrs()
+                parquet_file.cats = save_cats
+            else:
+                raise ValueError("Neither path nor ParquetFile detected!")
+
             if null_index_name:
-                if "__index_level_0__" in pf.columns:
+                if "__index_level_0__" in parquet_file.columns:
                     # See "Handling a None-labeled index" comment above
                     index = ["__index_level_0__"]
                     columns += index
-                    pf.fmd.key_value_metadata = None
-            else:
-                pf.fmd.key_value_metadata = None
-            return pf.read_row_group_file(
-                rg_piece, columns, categories, index=index, **kwargs.get("read", {})
-            )
+
+            parquet_file._dtypes = (
+                lambda *args: parquet_file.dtypes
+            )  # ugly patch, could be fixed
+
+            # Read necessary row-groups and concatenate
+            dfs = []
+            for row_group in row_groups:
+                dfs.append(
+                    parquet_file.read_row_group_file(
+                        row_group,
+                        columns,
+                        categories,
+                        index=index,
+                        **kwargs.get("read", {}),
+                    )
+                )
+            return concat(dfs, axis=0) if len(dfs) > 1 else dfs[0]
+
+        else:
+            # `piece` is NOT a tuple
+            raise ValueError(f"Expected tuple, got {type(piece)}")
 
     @classmethod
     def initialize_write(
@@ -518,7 +815,7 @@ class FastParquetEngine(Engine):
         ignore_divisions=False,
         division_info=None,
         schema=None,
-        **kwargs
+        **kwargs,
     ):
         if append and division_info is None:
             ignore_divisions = True
@@ -583,7 +880,7 @@ class FastParquetEngine(Engine):
                 object_encoding=object_encoding,
                 index_cols=index_cols,
                 ignore_columns=partition_on,
-                **kwargs
+                **kwargs,
             )
             i_offset = 0
 
@@ -601,7 +898,7 @@ class FastParquetEngine(Engine):
         return_metadata,
         fmd=None,
         compression=None,
-        **kwargs
+        **kwargs,
     ):
         fmd = copy.copy(fmd)
         if not len(df):
diff --git a/dask/dataframe/io/parquet/utils.py b/dask/dataframe/io/parquet/utils.py
index 8c164170e..1734590fe 100644
--- a/dask/dataframe/io/parquet/utils.py
+++ b/dask/dataframe/io/parquet/utils.py
@@ -1,5 +1,9 @@
 import re
 
+import pandas as pd
+
+from ....core import flatten
+
 
 class Engine:
     """ The API necessary to provide a new Parquet reader/writer """
@@ -54,8 +58,8 @@ class Engine:
 
             [
                 {'num-rows': 1000, 'columns': [
-                    {'name': 'id', 'min': 0, 'max': 100, 'null-count': 0},
-                    {'name': 'x', 'min': 0.0, 'max': 1.0, 'null-count': 5},
+                    {'name': 'id', 'min': 0, 'max': 100},
+                    {'name': 'x', 'min': 0.0, 'max': 1.0},
                     ]},
                 ...
             ]
@@ -476,3 +480,139 @@ def _analyze_paths(file_list, fs, root=False):
         "/".join(basepath),
         out_list,
     )  # use '/'.join() instead of _join_path to be consistent with split('/')
+
+
+def _flatten_filters(filters):
+    """Flatten DNF-formatted filters (list of tuples)"""
+    return (
+        set(flatten(tuple(flatten(filters, container=list)), container=tuple))
+        if filters
+        else []
+    )
+
+
+def _aggregate_stats(
+    file_path,
+    file_row_group_stats,
+    file_row_group_column_stats,
+    stat_col_indices,
+):
+    """Utility to aggregate the statistics for N row-groups
+    into a single dictionary.
+
+    Used by `Engine._construct_parts`
+    """
+    if len(file_row_group_stats) < 1:
+        # Empty statistics
+        return {}
+    elif len(file_row_group_column_stats) == 0:
+        assert len(file_row_group_stats) == 1
+        return file_row_group_stats[0]
+    else:
+        # Note: It would be better to avoid df_rgs and df_cols
+        #       construction altogether. It makes it fast to aggregate
+        #       the statistics for many row groups, but isn't
+        #       worthwhile for a small number of row groups.
+        if len(file_row_group_stats) > 1:
+            df_rgs = pd.DataFrame(file_row_group_stats)
+            s = {
+                "file_path_0": file_path,
+                "num-rows": df_rgs["num-rows"].sum(),
+                "total_byte_size": df_rgs["total_byte_size"].sum(),
+                "columns": [],
+            }
+        else:
+            s = {
+                "file_path_0": file_path,
+                "num-rows": file_row_group_stats[0]["num-rows"],
+                "total_byte_size": file_row_group_stats[0]["total_byte_size"],
+                "columns": [],
+            }
+
+        df_cols = None
+        if len(file_row_group_column_stats) > 1:
+            df_cols = pd.DataFrame(file_row_group_column_stats)
+        for ind, name in enumerate(stat_col_indices):
+            i = ind * 2
+            if df_cols is None:
+                s["columns"].append(
+                    {
+                        "name": name,
+                        "min": file_row_group_column_stats[0][i],
+                        "max": file_row_group_column_stats[0][i + 1],
+                    }
+                )
+            else:
+                s["columns"].append(
+                    {
+                        "name": name,
+                        "min": df_cols.iloc[:, i].min(),
+                        "max": df_cols.iloc[:, i + 1].max(),
+                    }
+                )
+        return s
+
+
+def _row_groups_to_parts(
+    gather_statistics,
+    split_row_groups,
+    file_row_groups,
+    file_row_group_stats,
+    file_row_group_column_stats,
+    stat_col_indices,
+    make_part_func,
+    make_part_kwargs,
+):
+
+    # Construct `parts` and `stats`
+    parts = []
+    stats = []
+    if split_row_groups:
+        # Create parts from each file,
+        # limiting the number of row_groups in each piece
+        split_row_groups = int(split_row_groups)
+        for filename, row_groups in file_row_groups.items():
+            row_group_count = len(row_groups)
+            for i in range(0, row_group_count, split_row_groups):
+                i_end = i + split_row_groups
+                rg_list = row_groups[i:i_end]
+
+                part = make_part_func(
+                    filename,
+                    rg_list,
+                    **make_part_kwargs,
+                )
+                if part is None:
+                    continue
+
+                parts.append(part)
+                if gather_statistics:
+                    stat = _aggregate_stats(
+                        filename,
+                        file_row_group_stats[filename][i:i_end],
+                        file_row_group_column_stats[filename][i:i_end],
+                        stat_col_indices,
+                    )
+                    stats.append(stat)
+    else:
+        for filename, row_groups in file_row_groups.items():
+
+            part = make_part_func(
+                filename,
+                row_groups,
+                **make_part_kwargs,
+            )
+            if part is None:
+                continue
+
+            parts.append(part)
+            if gather_statistics:
+                stat = _aggregate_stats(
+                    filename,
+                    file_row_group_stats[filename],
+                    file_row_group_column_stats[filename],
+                    stat_col_indices,
+                )
+                stats.append(stat)
+
+    return parts, stats
diff --git a/dask/dataframe/shuffle.py b/dask/dataframe/shuffle.py
index b86d928cf..3280215c2 100644
--- a/dask/dataframe/shuffle.py
+++ b/dask/dataframe/shuffle.py
@@ -159,7 +159,7 @@ class SimpleShuffleLayer(Layer):
         )
 
         if state["annotations"]:
-            annotations.update(cls.expand_annotations(state["annotations"], raw.keys()))
+            cls.unpack_annotations(annotations, state["annotations"], raw.keys())
 
     def _keys_to_parts(self, keys):
         """Simple utility to convert keys to partition indices."""
diff --git a/dask/highlevelgraph.py b/dask/highlevelgraph.py
index aa198f88b..d6f43d632 100644
--- a/dask/highlevelgraph.py
+++ b/dask/highlevelgraph.py
@@ -1,9 +1,11 @@
 import abc
 import collections.abc
+import warnings
 from typing import (
     Any,
     Dict,
     Hashable,
+    MutableMapping,
     Optional,
     Set,
     Mapping,
@@ -107,14 +109,25 @@ class Layer(collections.abc.Mapping):
         return packed
 
     @staticmethod
-    def expand_annotations(annotations, keys) -> Mapping[str, Any]:
-        if annotations is None:
+    def unpack_annotations(
+        annotations: MutableMapping[str, Any],
+        new_annotations: Mapping[str, Any],
+        keys: Iterable,
+    ) -> None:
+        """
+        Expand a set of layer annotations across a set of keys, then merge those
+        expanded annotations for the layer into an existing annotations mapping.
+        This is not a simple shallow merge because some annotations like retries,
+        priority, workers, etc need to be able to retain keys from different layers.
+        """
+        if new_annotations is None:
             return None
 
         expanded = {}
         keys_stringified = False
 
-        for a, v in annotations.items():
+        # Expand the new annotations across the keyset
+        for a, v in new_annotations.items():
             if type(v) is dict and "__expanded_annotations__" in v:
                 # Maybe do a destructive update for efficiency?
                 v = v.copy()
@@ -127,7 +140,14 @@ class Layer(collections.abc.Mapping):
 
                 expanded[a] = {k: v for k in keys}
 
-        return expanded
+        # Merge the expanded annotations with the existing annotations mapping
+        to_merge = {}
+        for k, v in expanded.items():
+            if isinstance(v, dict) and k in annotations:
+                to_merge[k] = {**annotations[k], **v}
+            else:
+                to_merge[k] = v
+        annotations.update(to_merge)
 
     def cull(
         self, keys: Set, all_hlg_keys: Iterable
@@ -226,7 +246,7 @@ class Layer(collections.abc.Mapping):
         cls,
         state: Any,
         dsk: Dict[str, Any],
-        dependencies: Mapping[Hashable, Set],
+        dependencies: MutableMapping[Hashable, Set],
         annotations: Dict[str, Any],
     ) -> None:
         """Unpack the state of a layer previously packed by __dask_distributed_pack__()
@@ -245,14 +265,19 @@ class Layer(collections.abc.Mapping):
             The state returned by Layer.__dask_distributed_pack__()
         dsk: dict
             The materialized low level graph of the already unpacked layers
-        dependencies: Mapping
+        dependencies: MutableMapping
             The dependencies of each key in `dsk`
         annotations: dict
             The materialized task annotations
         """
-        raise NotImplementedError(
-            f"{type(cls)} doesn't implement __dask_distributed_unpack__()"
-        )
+        dsk.update(state["dsk"])
+        for k, v in state["dependencies"].items():
+            dependencies[k] = set(dependencies.get(k, ())) | set(v)
+
+        if state["annotations"]:
+            cls.unpack_annotations(
+                annotations, state["annotations"], state["dsk"].keys()
+            )
 
     def __reduce__(self):
         """Default serialization implementation, which materializes the Layer
@@ -544,6 +569,12 @@ class HighLevelGraph(Mapping):
     @property
     def dicts(self):
         # Backwards compatibility for now
+        warnings.warn(
+            "'dicts' property of HighLevelGraph is deprecated now and will be "
+            "removed in a future version. To silence this warning, "
+            "use '.layers' instead.",
+            FutureWarning,
+        )
         return self.layers
 
     def items(self):
diff --git a/dask/utils.py b/dask/utils.py
index 871f8ead2..c38380934 100644
--- a/dask/utils.py
+++ b/dask/utils.py
@@ -1027,10 +1027,10 @@ def get_scheduler_lock(collection=None, scheduler=None):
 def ensure_dict(d):
     if type(d) is dict:
         return d
-    elif hasattr(d, "dicts"):
+    elif hasattr(d, "layers"):
         result = {}
         seen = set()
-        for dd in d.dicts.values():
+        for dd in d.layers.values():
             dd_id = id(dd)
             if dd_id not in seen:
                 result.update(dd)


Test patch:
diff --git a/dask/array/tests/test_atop.py b/dask/array/tests/test_atop.py
index 050765434..c409d8ce4 100644
--- a/dask/array/tests/test_atop.py
+++ b/dask/array/tests/test_atop.py
@@ -197,7 +197,7 @@ def test_optimize_blockwise():
     assert isinstance(dsk, HighLevelGraph)
 
     assert (
-        len([layer for layer in dsk.dicts.values() if isinstance(layer, Blockwise)])
+        len([layer for layer in dsk.layers.values() if isinstance(layer, Blockwise)])
         == 1
     )
 
@@ -220,7 +220,7 @@ def test_optimize_blockwise_annotations():
 
     annotations = (
         layer.annotations
-        for layer in dsk.dicts.values()
+        for layer in dsk.layers.values()
         if isinstance(layer, Blockwise)
     )
     annotations = collections.Counter(
@@ -245,7 +245,7 @@ def test_blockwise_diamond_fusion():
     assert isinstance(dsk, HighLevelGraph)
 
     assert (
-        len([layer for layer in dsk.dicts.values() if isinstance(layer, Blockwise)])
+        len([layer for layer in dsk.layers.values() if isinstance(layer, Blockwise)])
         == 1
     )
 
@@ -256,15 +256,15 @@ def test_blockwise_non_blockwise_output():
     w = y.sum()
     z = ((y * 2) * 3) * 4
 
-    z_top_before = tuple(z.dask.dicts[z.name].indices)
+    z_top_before = tuple(z.dask.layers[z.name].indices)
     (zz,) = dask.optimize(z)
-    z_top_after = tuple(z.dask.dicts[z.name].indices)
+    z_top_after = tuple(z.dask.layers[z.name].indices)
     assert z_top_before == z_top_after, "z_top mutated"
 
     dsk = optimize_blockwise(z.dask, keys=list(dask.core.flatten(z.__dask_keys__())))
     assert isinstance(dsk, HighLevelGraph)
     assert (
-        len([layer for layer in dsk.dicts.values() if isinstance(layer, Blockwise)])
+        len([layer for layer in dsk.layers.values() if isinstance(layer, Blockwise)])
         == 1
     )
 
@@ -274,7 +274,7 @@ def test_blockwise_non_blockwise_output():
     )
     assert isinstance(dsk, HighLevelGraph)
     assert (
-        len([layer for layer in z.dask.dicts.values() if isinstance(layer, Blockwise)])
+        len([layer for layer in z.dask.layers.values() if isinstance(layer, Blockwise)])
         >= 1
     )
 
@@ -283,7 +283,7 @@ def test_top_len():
     x = da.ones(10, chunks=(5,))
     y = x[:, None] * x[None, :]
 
-    d = y.dask.dicts[y.name]
+    d = y.dask.layers[y.name]
     assert len(d) == 4
 
 
@@ -606,7 +606,7 @@ def test_dont_merge_before_reductions():
 
     dsk = optimize_blockwise(w.dask)
 
-    assert len([d for d in dsk.dicts.values() if isinstance(d, Blockwise)]) == 2
+    assert len([d for d in dsk.layers.values() if isinstance(d, Blockwise)]) == 2
 
     z.compute()
 
diff --git a/dask/dataframe/io/tests/test_parquet.py b/dask/dataframe/io/tests/test_parquet.py
index c4552795e..f4d377e39 100644
--- a/dask/dataframe/io/tests/test_parquet.py
+++ b/dask/dataframe/io/tests/test_parquet.py
@@ -186,12 +186,7 @@ if (
     )
 else:
     fp_pandas_msg = "pandas with fastparquet engine does not preserve index"
-    fp_pandas_xfail = write_read_engines(
-        **{
-            "xfail_pyarrow-dataset_fastparquet": pyarrow_fastparquet_msg,
-            "xfail_pyarrow-legacy_fastparquet": pyarrow_fastparquet_msg,
-        }
-    )
+    fp_pandas_xfail = write_read_engines()
 
 
 @write_read_engines()
@@ -243,7 +238,10 @@ def test_empty(tmpdir, write_engine, read_engine, index):
 @write_read_engines()
 def test_simple(tmpdir, write_engine, read_engine):
     fn = str(tmpdir)
-    df = pd.DataFrame({"a": ["a", "b", "b"], "b": [4, 5, 6]})
+    if write_engine != "fastparquet":
+        df = pd.DataFrame({"a": [b"a", b"b", b"b"], "b": [4, 5, 6]})
+    else:
+        df = pd.DataFrame({"a": ["a", "b", "b"], "b": [4, 5, 6]})
     df.set_index("a", inplace=True, drop=True)
     ddf = dd.from_pandas(df, npartitions=2)
     ddf.to_parquet(fn, engine=write_engine)
@@ -1187,12 +1185,14 @@ def test_pyarrow_schema_inference(tmpdir, index, engine, schema):
     else:
         df = dd.from_pandas(df, npartitions=2)
 
-    df.to_parquet(tmpdir, engine="pyarrow", schema=schema, compute=False).compute(
-        scheduler="synchronous"
-    )
+    df.to_parquet(tmpdir, engine="pyarrow", schema=schema)
     df_out = dd.read_parquet(tmpdir, engine=engine)
+    df_out.compute()
 
     if index and engine == "fastparquet":
+        # Fastparquet fails to detect int64 from _metadata
+        df_out["amount"] = df_out["amount"].astype("int64")
+
         # Fastparquet not handling divisions for
         # pyarrow-written dataset with string index
         assert_eq(df, df_out, check_divisions=False)
@@ -2094,29 +2094,6 @@ def test_arrow_partitioning(tmpdir):
     ddf.astype({"b": np.float32}).compute()
 
 
-def test_sorted_warnings(tmpdir, engine):
-
-    if engine.startswith("pyarrow"):
-        pytest.skip(
-            "ArrowEngine will only collect statistics for "
-            "known index columns and/or filtered columns."
-        )
-
-    tmpdir = str(tmpdir)
-    df = dd.from_pandas(
-        pd.DataFrame({"cola": range(10), "colb": range(10)}), npartitions=2
-    )
-    df.to_parquet(tmpdir, engine=engine, write_index=False)
-    with pytest.warns(RuntimeWarning) as record:
-        out = dd.read_parquet(tmpdir, engine=engine)
-    assert "['cola', 'colb']" in str(record[-1].message)
-    warnings = len(record)
-    assert out.columns.tolist() == ["cola", "colb"]
-    with pytest.warns(None) as record:
-        dd.read_parquet(tmpdir, engine=engine, index=False)
-    assert len(record) < warnings  # still may have some arrow warnings
-
-
 def test_informative_error_messages():
     with pytest.raises(ValueError) as info:
         dd.read_parquet("foo", engine="foo")
@@ -2405,9 +2382,7 @@ def test_graph_size_pyarrow(tmpdir, engine):
     ddf1.to_parquet(fn, engine=engine)
     ddf2 = dd.read_parquet(fn, engine=engine)
 
-    # pyarrow.dataset requires fragments to be passed in dict.
-    # This requires slightly more space than "legacy"
-    assert len(pickle.dumps(ddf2.__dask_graph__())) < 32000
+    assert len(pickle.dumps(ddf2.__dask_graph__())) < 25000
 
 
 @pytest.mark.parametrize("preserve_index", [True, False])
@@ -2500,7 +2475,7 @@ def test_blockwise_parquet_annotations(tmpdir):
     assert list(layers.values())[0].annotations == {"foo": "bar"}
 
 
-def test_split_row_groups_pyarrow(tmpdir):
+def test_split_row_groups(tmpdir, engine):
     """Test split_row_groups read_parquet kwarg"""
     check_pyarrow()
     tmp = str(tmpdir)
@@ -2514,11 +2489,11 @@ def test_split_row_groups_pyarrow(tmpdir):
         tmp, engine="pyarrow", row_group_size=100
     )
 
-    ddf3 = dd.read_parquet(tmp, engine="pyarrow", split_row_groups=True, chunksize=1)
+    ddf3 = dd.read_parquet(tmp, engine=engine, split_row_groups=True, chunksize=1)
     assert ddf3.npartitions == 4
 
     ddf3 = dd.read_parquet(
-        tmp, engine="pyarrow", gather_statistics=True, split_row_groups=False
+        tmp, engine=engine, gather_statistics=True, split_row_groups=False
     )
     assert ddf3.npartitions == 2
 
@@ -2528,7 +2503,7 @@ def test_split_row_groups_pyarrow(tmpdir):
 
     ddf3 = dd.read_parquet(
         tmp,
-        engine="pyarrow",
+        engine=engine,
         gather_statistics=True,
         split_row_groups=True,
         chunksize=1,
@@ -2536,18 +2511,17 @@ def test_split_row_groups_pyarrow(tmpdir):
     assert ddf3.npartitions == 12
 
     ddf3 = dd.read_parquet(
-        tmp, engine="pyarrow", gather_statistics=True, split_row_groups=False
+        tmp, engine=engine, gather_statistics=True, split_row_groups=False
     )
     assert ddf3.npartitions == 4
 
 
 @pytest.mark.parametrize("split_row_groups", [1, 12])
 @pytest.mark.parametrize("gather_statistics", [True, False])
-def test_split_row_groups_int_pyarrow(tmpdir, split_row_groups, gather_statistics):
+def test_split_row_groups_int(tmpdir, split_row_groups, gather_statistics, engine):
 
     check_pyarrow()
     tmp = str(tmpdir)
-    engine = "pyarrow"
     row_group_size = 10
     npartitions = 4
     half_size = 400
@@ -2560,10 +2534,10 @@ def test_split_row_groups_int_pyarrow(tmpdir, split_row_groups, gather_statistic
     half = len(df) // 2
 
     dd.from_pandas(df.iloc[:half], npartitions=npartitions).to_parquet(
-        tmp, engine=engine, row_group_size=row_group_size
+        tmp, engine="pyarrow", row_group_size=row_group_size
     )
     dd.from_pandas(df.iloc[half:], npartitions=npartitions).to_parquet(
-        tmp, append=True, engine=engine, row_group_size=row_group_size
+        tmp, append=True, engine="pyarrow", row_group_size=row_group_size
     )
 
     ddf2 = dd.read_parquet(
@@ -2576,7 +2550,7 @@ def test_split_row_groups_int_pyarrow(tmpdir, split_row_groups, gather_statistic
     assert ddf2.npartitions == 2 * math.ceil(expected_rg_cout / split_row_groups)
 
 
-def test_split_row_groups_filter_pyarrow(tmpdir):
+def test_split_row_groups_filter(tmpdir, engine):
     check_pyarrow()
     tmp = str(tmpdir)
     df = pd.DataFrame(
@@ -2590,17 +2564,15 @@ def test_split_row_groups_filter_pyarrow(tmpdir):
         tmp, append=True, engine="pyarrow", row_group_size=50
     )
 
-    ddf2 = dd.read_parquet(tmp, engine="pyarrow")
+    ddf2 = dd.read_parquet(tmp, engine=engine)
     ddf3 = dd.read_parquet(
         tmp,
-        engine="pyarrow",
+        engine=engine,
         gather_statistics=True,
         split_row_groups=True,
         filters=filters,
     )
 
-    ddf2.compute(scheduler="sync")
-
     assert search_val in ddf3["i32"]
     assert_eq(
         ddf2[ddf2["i32"] == search_val].compute(),
diff --git a/dask/tests/test_distributed.py b/dask/tests/test_distributed.py
index 4bd3a25d9..24dcd6637 100644
--- a/dask/tests/test_distributed.py
+++ b/dask/tests/test_distributed.py
@@ -21,6 +21,7 @@ from distributed.utils_test import (  # noqa F401
     cluster_fixture,
     loop,
     client as c,
+    varying,
 )
 
 
@@ -213,3 +214,34 @@ def test_local_scheduler():
         assert len(z.dask) == 1
 
     asyncio.get_event_loop().run_until_complete(f())
+
+
+@gen_cluster(client=True)
+async def test_annotations_blockwise_unpack(c, s, a, b):
+    da = pytest.importorskip("dask.array")
+    np = pytest.importorskip("numpy")
+    from dask.array.utils import assert_eq
+
+    # A flaky doubling function -- need extra args because it is called before
+    # application to establish dtype/meta.
+    scale = varying([ZeroDivisionError("one"), ZeroDivisionError("two"), 2, 2])
+
+    def flaky_double(x):
+        return scale() * x
+
+    # A reliable double function.
+    def reliable_double(x):
+        return 2 * x
+
+    x = da.ones(10, chunks=(5,))
+
+    # The later annotations should not override the earlier annotations
+    with dask.annotate(retries=2):
+        y = x.map_blocks(flaky_double, meta=np.array((), dtype=np.float))
+    with dask.annotate(retries=0):
+        z = y.map_blocks(reliable_double, meta=np.array((), dtype=np.float))
+
+    with dask.config.set(optimization__fuse__active=False):
+        z = await c.compute(z)
+
+    assert_eq(z, np.ones(10) * 4.0)
diff --git a/dask/tests/test_highgraph.py b/dask/tests/test_highgraph.py
index ca5f2ce39..7f35248ff 100644
--- a/dask/tests/test_highgraph.py
+++ b/dask/tests/test_highgraph.py
@@ -130,3 +130,10 @@ def test_blockwise_cull(flat):
         out_keys = layer.get_output_keys()
         assert out_keys == {(layer.output, *select)}
         assert not layer.is_materialized()
+
+
+def test_highlevelgraph_dicts_deprecation():
+    with pytest.warns(FutureWarning):
+        layers = {"a": BasicLayer({"x": 1, "y": (inc, "x")})}
+        hg = HighLevelGraph(layers, {"a": set()})
+        assert hg.dicts == layers