Repository: dask/dask

Issue:
`_metadata` is re-read for every partition when using `fastparquet` engine
## Background
Given a (poorly) partitioned dataset of 15000 partitions of ~3k rows each, with a `_metadata` file, we evaluate e.g.

```python
df = dd.read_parquet(source, engine=engine)  # L1: mostly read _metadata
df.head()  # L2: single task
df.size.compute()  # L3: 15000 tasks
```

with engine set to either `fastparquet` or `pyarrow`. After [this speed-up](https://github.com/dask/dask/pull/5821), L1 takes ~25s with `fastparquet` vs 215s with `pyarrow`.

However, L2 takes ~20s with `fastparquet` vs ~200-300ms with `pyarrow`.

Similarly, L3 takes ~20s **per partition** (`read-parquet-...` task) on the workers - this appears to be largely due to repeated re-reading the metadata file, as confirmed by removing the `_metadata` file after executing L1.

This makes processing the data almost 100x slower than it could be, and leads to `O(n^2)` time in size of data when partitions are small.


Discussion:
rjzamora: @ig248 - Note that this PR addresses some of [your suggestions](https://github.com/dask/dask/issues/5842#issuecomment-590415160) from #5842 (although the current changes seem most helpful for non-partitioned datasets, since the `"_metadata"` file should no longer be re-read at all). 
mrocklin: cc @martindurant if you have a moment to review
martindurant: Is this still WIP, or are you ready for me to have a look, @rjzamora ?
rjzamora: Thanks @martindurant!  I am not planning to add anything here today, so a review at your leisure is certainly appreciated :)
martindurant: Failure is a RuntimeWarning on py38 in test_cov (array?) - perhaps a new compiler version of numpy? In other words, unrelated to this PR.
rjzamora: > Glad to see it passing! Do we have benchmarks?

Just a note: I will revisit this soonish, but I did not see significant performance improvements from these changes when I last checked.  For the partitioned dataset case, we no longer spend much time in `_determine_pf_parts`, but we still spend a lot of time creating `pf` for each partition.
martindurant: Can you please merge from master? I believe things should pass now.

Patch:
diff --git a/dask/array/creation.py b/dask/array/creation.py
index 84a589879..2ca9ddbfd 100644
--- a/dask/array/creation.py
+++ b/dask/array/creation.py
@@ -763,7 +763,9 @@ def repeat(a, repeats, axis=None):
     elif not 0 <= axis <= a.ndim - 1:
         raise ValueError("axis(=%d) out of bounds" % axis)
 
-    if repeats == 1:
+    if repeats == 0:
+        return a[tuple(slice(None) if d != axis else slice(0) for d in range(a.ndim))]
+    elif repeats == 1:
         return a
 
     cchunks = cached_cumsum(a.chunks[axis], initial_zero=True)
@@ -835,7 +837,7 @@ def expand_pad_value(array, pad_value):
         and len(pad_value) == 2
         and all(isinstance(pw, Number) for pw in pad_value)
     ):
-        pad_value = tuple((pad_value[0], pad_value[1]) for _ in range(array.ndim))
+        pad_value = array.ndim * (tuple(pad_value),)
     elif (
         isinstance(pad_value, Sequence)
         and len(pad_value) == array.ndim
@@ -843,7 +845,15 @@ def expand_pad_value(array, pad_value):
         and all((len(pw) == 2) for pw in pad_value)
         and all(all(isinstance(w, Number) for w in pw) for pw in pad_value)
     ):
-        pad_value = tuple((pw[0], pw[1]) for pw in pad_value)
+        pad_value = tuple(tuple(pw) for pw in pad_value)
+    elif (
+        isinstance(pad_value, Sequence)
+        and len(pad_value) == 1
+        and isinstance(pad_value[0], Sequence)
+        and len(pad_value[0]) == 2
+        and all(isinstance(pw, Number) for pw in pad_value[0])
+    ):
+        pad_value = array.ndim * (tuple(pad_value[0]),)
     else:
         raise TypeError("`pad_value` must be composed of integral typed values.")
 
@@ -1117,7 +1127,7 @@ def pad(array, pad_width, mode, **kwargs):
     pad_width = expand_pad_value(array, pad_width)
 
     if mode in ["maximum", "mean", "median", "minimum"]:
-        kwargs.setdefault("stat_length", array.shape)
+        kwargs.setdefault("stat_length", tuple((n, n) for n in array.shape))
     elif mode == "constant":
         kwargs.setdefault("constant_values", 0)
     elif mode == "linear_ramp":
diff --git a/dask/array/random.py b/dask/array/random.py
index ed19cfc4b..e4e445941 100644
--- a/dask/array/random.py
+++ b/dask/array/random.py
@@ -196,21 +196,21 @@ class RandomState(object):
         graph = HighLevelGraph.from_collections(name, dsk, dependencies=dependencies)
         return Array(graph, name, chunks + extra_chunks, meta=meta)
 
-    @derived_from(np.random.RandomState)
+    @derived_from(np.random.RandomState, skipblocks=1)
     def beta(self, a, b, size=None, chunks="auto", **kwargs):
         return self._wrap("beta", a, b, size=size, chunks=chunks, **kwargs)
 
-    @derived_from(np.random.RandomState)
+    @derived_from(np.random.RandomState, skipblocks=1)
     def binomial(self, n, p, size=None, chunks="auto", **kwargs):
         return self._wrap("binomial", n, p, size=size, chunks=chunks, **kwargs)
 
-    @derived_from(np.random.RandomState)
+    @derived_from(np.random.RandomState, skipblocks=1)
     def chisquare(self, df, size=None, chunks="auto", **kwargs):
         return self._wrap("chisquare", df, size=size, chunks=chunks, **kwargs)
 
     with ignoring(AttributeError):
 
-        @derived_from(np.random.RandomState)
+        @derived_from(np.random.RandomState, skipblocks=1)
         def choice(self, a, size=None, replace=True, p=None, chunks="auto"):
             dependencies = []
             # Normalize and validate `a`
@@ -282,52 +282,52 @@ class RandomState(object):
             )
             return Array(graph, name, chunks, dtype=dtype)
 
-    # @derived_from(np.random.RandomState)
+    # @derived_from(np.random.RandomState, skipblocks=1)
     # def dirichlet(self, alpha, size=None, chunks="auto"):
 
-    @derived_from(np.random.RandomState)
+    @derived_from(np.random.RandomState, skipblocks=1)
     def exponential(self, scale=1.0, size=None, chunks="auto", **kwargs):
         return self._wrap("exponential", scale, size=size, chunks=chunks, **kwargs)
 
-    @derived_from(np.random.RandomState)
+    @derived_from(np.random.RandomState, skipblocks=1)
     def f(self, dfnum, dfden, size=None, chunks="auto", **kwargs):
         return self._wrap("f", dfnum, dfden, size=size, chunks=chunks, **kwargs)
 
-    @derived_from(np.random.RandomState)
+    @derived_from(np.random.RandomState, skipblocks=1)
     def gamma(self, shape, scale=1.0, size=None, chunks="auto", **kwargs):
         return self._wrap("gamma", shape, scale, size=size, chunks=chunks, **kwargs)
 
-    @derived_from(np.random.RandomState)
+    @derived_from(np.random.RandomState, skipblocks=1)
     def geometric(self, p, size=None, chunks="auto", **kwargs):
         return self._wrap("geometric", p, size=size, chunks=chunks, **kwargs)
 
-    @derived_from(np.random.RandomState)
+    @derived_from(np.random.RandomState, skipblocks=1)
     def gumbel(self, loc=0.0, scale=1.0, size=None, chunks="auto", **kwargs):
         return self._wrap("gumbel", loc, scale, size=size, chunks=chunks, **kwargs)
 
-    @derived_from(np.random.RandomState)
+    @derived_from(np.random.RandomState, skipblocks=1)
     def hypergeometric(self, ngood, nbad, nsample, size=None, chunks="auto", **kwargs):
         return self._wrap(
             "hypergeometric", ngood, nbad, nsample, size=size, chunks=chunks, **kwargs
         )
 
-    @derived_from(np.random.RandomState)
+    @derived_from(np.random.RandomState, skipblocks=1)
     def laplace(self, loc=0.0, scale=1.0, size=None, chunks="auto", **kwargs):
         return self._wrap("laplace", loc, scale, size=size, chunks=chunks, **kwargs)
 
-    @derived_from(np.random.RandomState)
+    @derived_from(np.random.RandomState, skipblocks=1)
     def logistic(self, loc=0.0, scale=1.0, size=None, chunks="auto", **kwargs):
         return self._wrap("logistic", loc, scale, size=size, chunks=chunks, **kwargs)
 
-    @derived_from(np.random.RandomState)
+    @derived_from(np.random.RandomState, skipblocks=1)
     def lognormal(self, mean=0.0, sigma=1.0, size=None, chunks="auto", **kwargs):
         return self._wrap("lognormal", mean, sigma, size=size, chunks=chunks, **kwargs)
 
-    @derived_from(np.random.RandomState)
+    @derived_from(np.random.RandomState, skipblocks=1)
     def logseries(self, p, size=None, chunks="auto", **kwargs):
         return self._wrap("logseries", p, size=size, chunks=chunks, **kwargs)
 
-    @derived_from(np.random.RandomState)
+    @derived_from(np.random.RandomState, skipblocks=1)
     def multinomial(self, n, pvals, size=None, chunks="auto", **kwargs):
         return self._wrap(
             "multinomial",
@@ -338,31 +338,31 @@ class RandomState(object):
             extra_chunks=((len(pvals),),),
         )
 
-    @derived_from(np.random.RandomState)
+    @derived_from(np.random.RandomState, skipblocks=1)
     def negative_binomial(self, n, p, size=None, chunks="auto", **kwargs):
         return self._wrap("negative_binomial", n, p, size=size, chunks=chunks, **kwargs)
 
-    @derived_from(np.random.RandomState)
+    @derived_from(np.random.RandomState, skipblocks=1)
     def noncentral_chisquare(self, df, nonc, size=None, chunks="auto", **kwargs):
         return self._wrap(
             "noncentral_chisquare", df, nonc, size=size, chunks=chunks, **kwargs
         )
 
-    @derived_from(np.random.RandomState)
+    @derived_from(np.random.RandomState, skipblocks=1)
     def noncentral_f(self, dfnum, dfden, nonc, size=None, chunks="auto", **kwargs):
         return self._wrap(
             "noncentral_f", dfnum, dfden, nonc, size=size, chunks=chunks, **kwargs
         )
 
-    @derived_from(np.random.RandomState)
+    @derived_from(np.random.RandomState, skipblocks=1)
     def normal(self, loc=0.0, scale=1.0, size=None, chunks="auto", **kwargs):
         return self._wrap("normal", loc, scale, size=size, chunks=chunks, **kwargs)
 
-    @derived_from(np.random.RandomState)
+    @derived_from(np.random.RandomState, skipblocks=1)
     def pareto(self, a, size=None, chunks="auto", **kwargs):
         return self._wrap("pareto", a, size=size, chunks=chunks, **kwargs)
 
-    @derived_from(np.random.RandomState)
+    @derived_from(np.random.RandomState, skipblocks=1)
     def permutation(self, x):
         from .slicing import shuffle_slice
 
@@ -373,83 +373,83 @@ class RandomState(object):
         self._numpy_state.shuffle(index)
         return shuffle_slice(x, index)
 
-    @derived_from(np.random.RandomState)
+    @derived_from(np.random.RandomState, skipblocks=1)
     def poisson(self, lam=1.0, size=None, chunks="auto", **kwargs):
         return self._wrap("poisson", lam, size=size, chunks=chunks, **kwargs)
 
-    @derived_from(np.random.RandomState)
+    @derived_from(np.random.RandomState, skipblocks=1)
     def power(self, a, size=None, chunks="auto", **kwargs):
         return self._wrap("power", a, size=size, chunks=chunks, **kwargs)
 
-    @derived_from(np.random.RandomState)
+    @derived_from(np.random.RandomState, skipblocks=1)
     def randint(self, low, high=None, size=None, chunks="auto", dtype="l", **kwargs):
         return self._wrap(
             "randint", low, high, size=size, chunks=chunks, dtype=dtype, **kwargs
         )
 
-    @derived_from(np.random.RandomState)
+    @derived_from(np.random.RandomState, skipblocks=1)
     def random_integers(self, low, high=None, size=None, chunks="auto", **kwargs):
         return self._wrap(
             "random_integers", low, high, size=size, chunks=chunks, **kwargs
         )
 
-    @derived_from(np.random.RandomState)
+    @derived_from(np.random.RandomState, skipblocks=1)
     def random_sample(self, size=None, chunks="auto", **kwargs):
         return self._wrap("random_sample", size=size, chunks=chunks, **kwargs)
 
     random = random_sample
 
-    @derived_from(np.random.RandomState)
+    @derived_from(np.random.RandomState, skipblocks=1)
     def rayleigh(self, scale=1.0, size=None, chunks="auto", **kwargs):
         return self._wrap("rayleigh", scale, size=size, chunks=chunks, **kwargs)
 
-    @derived_from(np.random.RandomState)
+    @derived_from(np.random.RandomState, skipblocks=1)
     def standard_cauchy(self, size=None, chunks="auto", **kwargs):
         return self._wrap("standard_cauchy", size=size, chunks=chunks, **kwargs)
 
-    @derived_from(np.random.RandomState)
+    @derived_from(np.random.RandomState, skipblocks=1)
     def standard_exponential(self, size=None, chunks="auto", **kwargs):
         return self._wrap("standard_exponential", size=size, chunks=chunks, **kwargs)
 
-    @derived_from(np.random.RandomState)
+    @derived_from(np.random.RandomState, skipblocks=1)
     def standard_gamma(self, shape, size=None, chunks="auto", **kwargs):
         return self._wrap("standard_gamma", shape, size=size, chunks=chunks, **kwargs)
 
-    @derived_from(np.random.RandomState)
+    @derived_from(np.random.RandomState, skipblocks=1)
     def standard_normal(self, size=None, chunks="auto", **kwargs):
         return self._wrap("standard_normal", size=size, chunks=chunks, **kwargs)
 
-    @derived_from(np.random.RandomState)
+    @derived_from(np.random.RandomState, skipblocks=1)
     def standard_t(self, df, size=None, chunks="auto", **kwargs):
         return self._wrap("standard_t", df, size=size, chunks=chunks, **kwargs)
 
-    @derived_from(np.random.RandomState)
+    @derived_from(np.random.RandomState, skipblocks=1)
     def tomaxint(self, size=None, chunks="auto", **kwargs):
         return self._wrap("tomaxint", size=size, chunks=chunks, **kwargs)
 
-    @derived_from(np.random.RandomState)
+    @derived_from(np.random.RandomState, skipblocks=1)
     def triangular(self, left, mode, right, size=None, chunks="auto", **kwargs):
         return self._wrap(
             "triangular", left, mode, right, size=size, chunks=chunks, **kwargs
         )
 
-    @derived_from(np.random.RandomState)
+    @derived_from(np.random.RandomState, skipblocks=1)
     def uniform(self, low=0.0, high=1.0, size=None, chunks="auto", **kwargs):
         return self._wrap("uniform", low, high, size=size, chunks=chunks, **kwargs)
 
-    @derived_from(np.random.RandomState)
+    @derived_from(np.random.RandomState, skipblocks=1)
     def vonmises(self, mu, kappa, size=None, chunks="auto", **kwargs):
         return self._wrap("vonmises", mu, kappa, size=size, chunks=chunks, **kwargs)
 
-    @derived_from(np.random.RandomState)
+    @derived_from(np.random.RandomState, skipblocks=1)
     def wald(self, mean, scale, size=None, chunks="auto", **kwargs):
         return self._wrap("wald", mean, scale, size=size, chunks=chunks, **kwargs)
 
-    @derived_from(np.random.RandomState)
+    @derived_from(np.random.RandomState, skipblocks=1)
     def weibull(self, a, size=None, chunks="auto", **kwargs):
         return self._wrap("weibull", a, size=size, chunks=chunks, **kwargs)
 
-    @derived_from(np.random.RandomState)
+    @derived_from(np.random.RandomState, skipblocks=1)
     def zipf(self, a, size=None, chunks="auto", **kwargs):
         return self._wrap("zipf", a, size=size, chunks=chunks, **kwargs)
 
diff --git a/dask/config.py b/dask/config.py
index a8c971b0c..415efd695 100644
--- a/dask/config.py
+++ b/dask/config.py
@@ -1,9 +1,10 @@
 import ast
 import builtins
+from collections.abc import Mapping
 import os
 import sys
 import threading
-from collections.abc import Mapping
+import warnings
 
 try:
     import yaml
@@ -300,10 +301,13 @@ class set(object):
 
             if arg is not None:
                 for key, value in arg.items():
+                    key = check_deprecations(key)
                     self._assign(key.split("."), value, config)
             if kwargs:
                 for key, value in kwargs.items():
-                    self._assign(key.split("__"), value, config)
+                    key = key.replace("__", ".")
+                    key = check_deprecations(key)
+                    self._assign(key.split("."), value, config)
 
     def __enter__(self):
         return self.config
@@ -341,6 +345,7 @@ class set(object):
             Whether this operation needs to be recorded to allow for rollback.
         """
         key = canonical_name(keys[0], d)
+
         path = path + (key,)
 
         if len(keys) == 1:
@@ -522,6 +527,60 @@ def expand_environment_variables(config):
         return config
 
 
+deprecations = {
+    "fuse_ave_width": "optimization.fuse.ave-width",
+    "fuse_max_height": "optimization.fuse.max-height",
+    "fuse_max_width": "optimization.fuse.max-width",
+    "fuse_subgraphs": "optimization.fuse.subgraphs",
+    "fuse_rename_keys": "optimization.fuse.rename-keys",
+    "fuse_max_depth_new_edges": "optimization.fuse.max-depth-new-edges",
+}
+
+
+def check_deprecations(key: str, deprecations: dict = deprecations):
+    """ Check if the provided value has been renamed or removed
+
+    Parameters
+    ----------
+    key : str
+        The configuration key to check
+    deprecations : Dict[str, str]
+        The mapping of aliases
+
+    Examples
+    --------
+    >>> deprecations = {"old_key": "new_key", "invalid": None}
+    >>> check_deprecations("old_key", deprecations=deprecations)  # doctest: +SKIP
+    UserWarning: Configuration key "old_key" has been deprecated. Please use "new_key" instead.
+
+    >>> check_deprecations("invalid", deprecations=deprecations)
+    Traceback (most recent call last):
+        ...
+    ValueError: Configuration value "invalid" has been removed
+
+    >>> check_deprecations("another_key", deprecations=deprecations)
+    'another_key'
+
+    Returns
+    -------
+    new: str
+        The proper key, whether the original (if no deprecation) or the aliased
+        value
+    """
+    if key in deprecations:
+        new = deprecations[key]
+        if new:
+            warnings.warn(
+                'Configuration key "{}" has been deprecated. '
+                'Please use "{}" instead'.format(key, new)
+            )
+            return new
+        else:
+            raise ValueError('Configuration value "{}" has been removed'.format(key))
+    else:
+        return key
+
+
 refresh()
 
 
diff --git a/dask/dask.yaml b/dask/dask.yaml
index d95314e01..3c881ac9a 100644
--- a/dask/dask.yaml
+++ b/dask/dask.yaml
@@ -6,3 +6,10 @@ dataframe:
 array:
   svg:
     size: 120  # pixels
+
+optimization:
+  fuse:
+    active: true
+    ave-width: 1
+    subraphs: false
+    rename-keys: true
diff --git a/dask/dataframe/core.py b/dask/dataframe/core.py
index 98c984b1a..9bb08d69e 100644
--- a/dask/dataframe/core.py
+++ b/dask/dataframe/core.py
@@ -58,6 +58,7 @@ from .utils import (
     insert_meta_param_description,
     raise_on_meta_error,
     clear_known_categories,
+    group_split_dispatch,
     is_categorical_dtype,
     has_known_categories,
     PANDAS_VERSION,
@@ -498,7 +499,9 @@ Dask Name: {name}, {task} tasks"""
             raise ValueError(msg)
 
     @derived_from(pd.DataFrame)
-    def drop_duplicates(self, subset=None, split_every=None, split_out=1, **kwargs):
+    def drop_duplicates(
+        self, subset=None, split_every=None, split_out=1, ignore_index=False, **kwargs
+    ):
         if subset is not None:
             # Let pandas error on bad inputs
             self._meta_nonempty.drop_duplicates(subset=subset, **kwargs)
@@ -523,6 +526,7 @@ Dask Name: {name}, {task} tasks"""
             split_out=split_out,
             split_out_setup=split_out_setup,
             split_out_setup_kwargs=split_out_setup_kwargs,
+            ignore_index=ignore_index,
             **kwargs
         )
 
@@ -627,7 +631,7 @@ Dask Name: {name}, {task} tasks"""
         In the case where the metadata doesn't change, you can also pass in
         the object itself directly:
 
-        >>> res = ddf.map_partitions(lambda df: df.head(), meta=df)
+        >>> res = ddf.map_partitions(lambda df: df.head(), meta=ddf)
 
         Also note that the index and divisions are assumed to remain unchanged.
         If the function you're mapping changes the index/divisions, you'll need
@@ -3359,7 +3363,7 @@ class DataFrame(_Frame):
 
     def __len__(self):
         try:
-            s = self[self.columns[0]]
+            s = self.iloc[:, 0]
         except IndexError:
             return super().__len__()
         else:
@@ -4716,7 +4720,9 @@ def _maybe_from_pandas(dfs):
     return dfs
 
 
-def hash_shard(df, nparts, split_out_setup=None, split_out_setup_kwargs=None):
+def hash_shard(
+    df, nparts, split_out_setup=None, split_out_setup_kwargs=None, ignore_index=False
+):
     if split_out_setup:
         h = split_out_setup(df, **(split_out_setup_kwargs or {}))
     else:
@@ -4725,8 +4731,8 @@ def hash_shard(df, nparts, split_out_setup=None, split_out_setup_kwargs=None):
     h = hash_object_dispatch(h, index=False)
     if is_series_like(h):
         h = h.values
-    h %= nparts
-    return {i: df.iloc[h == i] for i in range(nparts)}
+    np.mod(h, nparts, out=h)
+    return group_split_dispatch(df, h, nparts, ignore_index=ignore_index)
 
 
 def split_evenly(df, k):
@@ -4762,6 +4768,7 @@ def apply_concat_apply(
     split_out_setup=None,
     split_out_setup_kwargs=None,
     sort=None,
+    ignore_index=False,
     **kwargs
 ):
     """Apply a function to blocks, then concat, then apply again
@@ -4803,6 +4810,8 @@ def apply_concat_apply(
         Keywords for the `split_out_setup` function only.
     sort : bool, default None
         If allowed, sort the keys of the output aggregation.
+    ignore_index : bool, default False
+        If True, do not preserve index values throughout ACA operations.
     kwargs :
         All remaining keywords will be passed to ``chunk``, ``aggregate``, and
         ``combine``.
@@ -4892,6 +4901,7 @@ def apply_concat_apply(
                 split_out,
                 split_out_setup,
                 split_out_setup_kwargs,
+                ignore_index,
             )
             for j in range(split_out):
                 dsk[(shard_prefix, 0, i, j)] = (getitem, (split_prefix, i), j)
@@ -4906,7 +4916,7 @@ def apply_concat_apply(
     while k > split_every:
         for part_i, inds in enumerate(partition_all(split_every, range(k))):
             for j in range(split_out):
-                conc = (_concat, [(a, depth, i, j) for i in inds])
+                conc = (_concat, [(a, depth, i, j) for i in inds], ignore_index)
                 if combine_kwargs:
                     dsk[(b, depth + 1, part_i, j)] = (
                         apply,
@@ -4932,7 +4942,7 @@ def apply_concat_apply(
     # Aggregate
     for j in range(split_out):
         b = "{0}-agg-{1}".format(token or funcname(aggregate), token_key)
-        conc = (_concat, [(a, depth, i, j) for i in range(k)])
+        conc = (_concat, [(a, depth, i, j) for i in range(k)], ignore_index)
         if aggregate_kwargs:
             dsk[(b, j)] = (apply, aggregate, [conc], aggregate_kwargs)
         else:
@@ -4940,7 +4950,9 @@ def apply_concat_apply(
 
     if meta is no_default:
         meta_chunk = _emulate(chunk, *args, udf=True, **chunk_kwargs)
-        meta = _emulate(aggregate, _concat([meta_chunk]), udf=True, **aggregate_kwargs)
+        meta = _emulate(
+            aggregate, _concat([meta_chunk], ignore_index), udf=True, **aggregate_kwargs
+        )
     meta = make_meta(
         meta, index=(getattr(make_meta(dfs[0]), "index", None) if dfs else None)
     )
diff --git a/dask/dataframe/io/parquet/arrow.py b/dask/dataframe/io/parquet/arrow.py
index b75a764d0..2fae52a1e 100644
--- a/dask/dataframe/io/parquet/arrow.py
+++ b/dask/dataframe/io/parquet/arrow.py
@@ -546,9 +546,34 @@ class ArrowEngine(Engine):
             use_pandas_metadata=True,
             use_threads=False,
             **kwargs.get("read", {}),
-        ).to_pandas(categories=categories, use_threads=False, ignore_metadata=True)[
-            list(columns_and_parts)
-        ]
+        ).to_pandas(categories=categories, use_threads=False, ignore_metadata=False)
+
+        # Note that `to_pandas(ignore_metadata=False)` means
+        # pyarrow will use the pandas metadata to set the index.
+        index_in_columns_and_parts = set(df.index.names).issubset(
+            set(columns_and_parts)
+        )
+        if not index:
+            if index_in_columns_and_parts:
+                # User does not want to set index and a desired
+                # column/partition has been set to the index
+                df.reset_index(drop=False, inplace=True)
+            else:
+                # User does not want to set index and an
+                # "unwanted" column has been set to the index
+                df.reset_index(drop=True, inplace=True)
+        else:
+            if set(df.index.names) != set(index) and index_in_columns_and_parts:
+                # The wrong index has been set and it contains
+                # one or more desired columns/partitions
+                df.reset_index(drop=False, inplace=True)
+            elif index_in_columns_and_parts:
+                # The correct index has already been set
+                index = False
+                columns_and_parts = list(
+                    set(columns_and_parts).difference(set(df.index.names))
+                )
+        df = df[list(columns_and_parts)]
 
         if index:
             df = df.set_index(index)
diff --git a/dask/dataframe/io/parquet/fastparquet.py b/dask/dataframe/io/parquet/fastparquet.py
index 6ef9534e9..7b97f8745 100644
--- a/dask/dataframe/io/parquet/fastparquet.py
+++ b/dask/dataframe/io/parquet/fastparquet.py
@@ -119,11 +119,14 @@ def _determine_pf_parts(fs, paths, gather_statistics, **kwargs):
         if gather_statistics is not False:
             # This scans all the files, allowing index/divisions
             # and filtering
-            pf = ParquetFile(
-                paths, open_with=fs.open, sep=fs.sep, **kwargs.get("file", {})
-            )
             if "_metadata" not in fns:
+                paths_use = paths
                 fast_metadata = False
+            else:
+                paths_use = base + fs.sep + "_metadata"
+            pf = ParquetFile(
+                paths_use, open_with=fs.open, sep=fs.sep, **kwargs.get("file", {})
+            )
         else:
             if "_metadata" in fns:
                 # We have a _metadata file, lets use it
@@ -177,11 +180,12 @@ def _determine_pf_parts(fs, paths, gather_statistics, **kwargs):
             parts = paths.copy()
     else:
         # There is only one file to read
+        base = None
         pf = ParquetFile(
             paths[0], open_with=fs.open, sep=fs.sep, **kwargs.get("file", {})
         )
 
-    return parts, pf, gather_statistics, fast_metadata
+    return parts, pf, gather_statistics, fast_metadata, base
 
 
 class FastParquetEngine(Engine):
@@ -199,7 +203,7 @@ class FastParquetEngine(Engine):
         # 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 = _determine_pf_parts(
+        parts, pf, gather_statistics, fast_metadata, base_path = _determine_pf_parts(
             fs, paths, gather_statistics, **kwargs
         )
 
@@ -284,7 +288,6 @@ 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:
@@ -341,28 +344,97 @@ class FastParquetEngine(Engine):
         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
-        if not parts:
-            partsin = pf.row_groups
-            if fast_metadata:
-                pf = (paths, gather_statistics)
-        else:
+        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
-            pf = None
+        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 = []
-        for i, piece in enumerate(partsin):
-            if pf and not fast_metadata:
-                for col in piece.columns:
+        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
+            ):
+                # 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
+                else:
+                    i_path = 0
+                    path_last = path_curr
+                file_path = base_path + path_curr
+            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
-            piece_item = i if pf else piece
-            part = {
-                "piece": piece_item,
-                "kwargs": {"pf": pf, "categories": categories_dict or categories},
+
+            # 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, "categories": categories_dict or categories},
             }
-            parts.append(part)
+            parts.append(part_item)
 
         return (meta, stats, parts)
 
@@ -385,7 +457,12 @@ class FastParquetEngine(Engine):
             return pf.to_pandas(columns, categories, index=index)
         else:
             if isinstance(pf, tuple):
-                pf = _determine_pf_parts(fs, pf[0], pf[1], **kwargs)[1]
+                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]
diff --git a/dask/dataframe/optimize.py b/dask/dataframe/optimize.py
index 3e55f491c..090e1fd92 100644
--- a/dask/dataframe/optimize.py
+++ b/dask/dataframe/optimize.py
@@ -29,7 +29,7 @@ def optimize(dsk, keys, **kwargs):
         dsk,
         keys,
         dependencies=dependencies,
-        fuse_subgraphs=config.get("fuse_subgraphs", True),
+        fuse_subgraphs=config.get("optimization.fuse.subgraphs", True),
     )
     dsk, _ = cull(dsk, keys)
     return dsk
diff --git a/dask/dataframe/shuffle.py b/dask/dataframe/shuffle.py
index 4553aa227..c2540b557 100644
--- a/dask/dataframe/shuffle.py
+++ b/dask/dataframe/shuffle.py
@@ -674,11 +674,9 @@ def set_partitions_pre(s, divisions):
 def shuffle_group_2(df, col, ignore_index):
     if not len(df):
         return {}, df
-    ind = df[col].astype(np.int64)
+    ind = df[col].astype(np.int32)
     n = ind.max() + 1
-    result2 = group_split_dispatch(
-        df, ind.values.view(np.int64), n, ignore_index=ignore_index
-    )
+    result2 = group_split_dispatch(df, ind.values.view(), n, ignore_index=ignore_index)
     return result2, df.iloc[:0]
 
 
@@ -728,7 +726,7 @@ def shuffle_group(df, col, stage, k, npartitions, ignore_index):
     np.floor_divide(c, k ** stage, out=c)
     np.mod(c, k, out=c)
 
-    return group_split_dispatch(df, c.astype(np.int64), k, ignore_index=ignore_index)
+    return group_split_dispatch(df, c, k, ignore_index=ignore_index)
 
 
 @contextlib.contextmanager
diff --git a/dask/dataframe/utils.py b/dask/dataframe/utils.py
index ae12096dc..6b07f5a80 100644
--- a/dask/dataframe/utils.py
+++ b/dask/dataframe/utils.py
@@ -478,7 +478,9 @@ group_split_dispatch = Dispatch("group_split_dispatch")
 
 @group_split_dispatch.register((pd.DataFrame, pd.Series, pd.Index))
 def group_split_pandas(df, c, k, ignore_index=False):
-    indexer, locations = pd._libs.algos.groupsort_indexer(c, k)
+    indexer, locations = pd._libs.algos.groupsort_indexer(
+        c.astype(np.int64, copy=False), k
+    )
     df2 = df.take(indexer)
     locations = locations.cumsum()
     parts = [df2.iloc[a:b] for a, b in zip(locations[:-1], locations[1:])]
diff --git a/dask/optimization.py b/dask/optimization.py
index 7a00ffc7e..eafc9eb87 100644
--- a/dask/optimization.py
+++ b/dask/optimization.py
@@ -377,25 +377,41 @@ def functions_of(task):
     return funcs
 
 
-def default_fused_keys_renamer(keys):
-    """Create new keys for ``fuse`` tasks"""
+def default_fused_keys_renamer(keys, max_fused_key_length=120):
+    """Create new keys for ``fuse`` tasks.
+
+    The optional parameter `max_fused_key_length` is used to limit the maximum string length for each renamed key.
+    If this parameter is set to `None`, there is no limit.
+    """
     it = reversed(keys)
     first_key = next(it)
     typ = type(first_key)
+
+    if max_fused_key_length:  # Take into account size of hash suffix
+        max_fused_key_length -= 5
+
+    def _enforce_max_key_limit(key_name):
+        if max_fused_key_length and len(key_name) > max_fused_key_length:
+            name_hash = f"{hash(key_name):x}"[:4]
+            key_name = f"{key_name[:max_fused_key_length]}-{name_hash}"
+        return key_name
+
     if typ is str:
         first_name = key_split(first_key)
         names = {key_split(k) for k in it}
         names.discard(first_name)
         names = sorted(names)
         names.append(first_key)
-        return "-".join(names)
+        concatenated_name = "-".join(names)
+        return _enforce_max_key_limit(concatenated_name)
     elif typ is tuple and len(first_key) > 0 and isinstance(first_key[0], str):
         first_name = key_split(first_key)
         names = {key_split(k) for k in it}
         names.discard(first_name)
         names = sorted(names)
         names.append(first_key[0])
-        return ("-".join(names),) + first_key[1:]
+        concatenated_name = "-".join(names)
+        return (_enforce_max_key_limit(concatenated_name),) + first_key[1:]
 
 
 def fuse(
@@ -455,6 +471,9 @@ def fuse(
     dependencies: dict mapping dependencies after fusion.  Useful side effect
         to accelerate other downstream optimizations.
     """
+    if not config.get("optimization.fuse.active", True):
+        return dsk, dependencies
+
     if keys is not None and not isinstance(keys, set):
         if not isinstance(keys, list):
             keys = [keys]
@@ -462,36 +481,26 @@ def fuse(
 
     # Assign reasonable, not too restrictive defaults
     if ave_width is None:
-        if config.get("fuse_ave_width", None) is None:
-            ave_width = 1
-        else:
-            ave_width = config.get("fuse_ave_width", None)
-
+        ave_width = config.get("optimization.fuse.ave-width", 1)
     if max_height is None:
-        if config.get("fuse_max_height", None) is None:
-            max_height = len(dsk)
-        else:
-            max_height = config.get("fuse_max_height", None)
-
+        max_height = config.get("optimization.fuse.max-height", None) or len(dsk)
     max_depth_new_edges = (
         max_depth_new_edges
-        or config.get("fuse_max_depth_new_edges", None)
-        or ave_width + 1.5
+        or config.get("optimization.fuse.max-depth-new-edges", None)
+        or ave_width * 1.5
     )
     max_width = (
         max_width
-        or config.get("fuse_max_width", None)
+        or config.get("optimization.fuse.max-width", None)
         or 1.5 + ave_width * math.log(ave_width + 1)
     )
-
-    if fuse_subgraphs is None:
-        fuse_subgraphs = config.get("fuse_subgraphs", False)
+    fuse_subgraphs = fuse_subgraphs or config.get("optimization.fuse.subgraphs", False)
 
     if not ave_width or not max_height:
         return dsk, dependencies
 
     if rename_keys is None:
-        rename_keys = config.get("fuse_rename_keys", True)
+        rename_keys = config.get("optimization.fuse.rename-keys", True)
     if rename_keys is True:
         key_renamer = default_fused_keys_renamer
     elif rename_keys is False:
diff --git a/dask/utils.py b/dask/utils.py
index 9d33b4369..bcbac89d9 100644
--- a/dask/utils.py
+++ b/dask/utils.py
@@ -565,7 +565,7 @@ def extra_titles(doc):
     return "\n".join(lines)
 
 
-def ignore_warning(doc, cls, name, extra=""):
+def ignore_warning(doc, cls, name, extra="", skipblocks=0):
     """Expand docstring by adding disclaimer and extra text"""
     import inspect
 
@@ -584,6 +584,11 @@ def ignore_warning(doc, cls, name, extra=""):
         # Insert our warning
         head = doc[: i + 2]
         tail = doc[i + 2 :]
+        while skipblocks > 0:
+            i = tail.find("\n\n")
+            head = tail[: i + 2]
+            tail = tail[i + 2 :]
+            skipblocks -= 1
         # Indentation of next line
         indent = re.match(r"\s*", tail).group(0)
         # Insert the warning, indented, with a blank line before and after
@@ -612,7 +617,7 @@ def unsupported_arguments(doc, args):
     return "\n".join(lines)
 
 
-def _derived_from(cls, method, ua_args=[], extra=""):
+def _derived_from(cls, method, ua_args=[], extra="", skipblocks=0):
     """ Helper function for derived_from to ease testing """
     # do not use wraps here, as it hides keyword arguments displayed
     # in the doc
@@ -628,7 +633,9 @@ def _derived_from(cls, method, ua_args=[], extra=""):
 
     # Insert disclaimer that this is a copied docstring
     if doc:
-        doc = ignore_warning(doc, cls, method.__name__, extra=extra)
+        doc = ignore_warning(
+            doc, cls, method.__name__, extra=extra, skipblocks=skipblocks
+        )
     elif extra:
         doc += extra.rstrip("\n") + "\n\n"
 
@@ -650,7 +657,7 @@ def _derived_from(cls, method, ua_args=[], extra=""):
     return doc
 
 
-def derived_from(original_klass, version=None, ua_args=[]):
+def derived_from(original_klass, version=None, ua_args=[], skipblocks=0):
     """Decorator to attach original class's docstring to the wrapped method.
 
     The output structure will be: top line of docstring, disclaimer about this
@@ -667,13 +674,20 @@ def derived_from(original_klass, version=None, ua_args=[]):
     ua_args : list
         List of keywords which Dask doesn't support. Keywords existing in
         original but not in Dask will automatically be added.
+    skipblocks : int
+        How many text blocks (paragraphs) to skip from the start of the
+        docstring. Useful for cases where the target has extra front-matter.
     """
 
     def wrapper(method):
         try:
             extra = getattr(method, "__doc__", None) or ""
             method.__doc__ = _derived_from(
-                original_klass, method, ua_args=ua_args, extra=extra
+                original_klass,
+                method,
+                ua_args=ua_args,
+                extra=extra,
+                skipblocks=skipblocks,
             )
             return method
 
diff --git a/docs/source/delayed.rst b/docs/source/delayed.rst
index 1ba872d3b..889ba7f7d 100644
--- a/docs/source/delayed.rst
+++ b/docs/source/delayed.rst
@@ -51,7 +51,7 @@ abstractions like Dask Array or Dask DataFrame.  Consider the following example:
         return x + 1
 
     def double(x):
-        return x + 2
+        return x * 2
 
     def add(x, y):
         return x + y
@@ -138,7 +138,7 @@ reproduction of our original problem as a parallel code:
 
     @dask.delayed
     def double(x):
-        return x + 2
+        return x * 2
 
     @dask.delayed
     def add(x, y):
diff --git a/docs/source/setup/custom-startup.rst b/docs/source/setup/custom-startup.rst
index ccf938832..f3636cf19 100644
--- a/docs/source/setup/custom-startup.rst
+++ b/docs/source/setup/custom-startup.rst
@@ -26,6 +26,8 @@ initialization.
 
 .. _Click: http://click.pocoo.org/
 
+Example
+~~~~~~~
 
 As an example, consider the following file that creates a
 `scheduler plugin <https://distributed.dask.org/en/latest/plugins.html>`_
@@ -59,6 +61,49 @@ if it is on the path) when we start the scheduler::
 
    dask-scheduler --preload scheduler-setup.py --print-count
 
+Types
+~~~~~
+
+Preloads can be specified as any of the following forms:
+
+-   A path to a script, like ``/path/to/myfile.py``
+-   A module name that is on the path, like ``my_module.initialize``
+-   The text of a Python script, like ``import os; os.environ["A"] = "value"``
+
+Configuration
+~~~~~~~~~~~~~
+
+Preloads can also be registered with configuration at the following values:
+
+.. code-block:: yaml
+
+   distributed:
+     scheduler:
+       preload:
+       - "import os; os.environ['A'] = 'b'"  # use Python text
+       - /path/to/myfile.py                  # or a filename
+       - my_module                           # or a module name
+       preload_argv:
+       - []                                  # Pass optional keywords
+       - ["--option", "value"]
+       - []
+     worker:
+       preload: []
+       preload_argv: []
+     nanny:
+       preload: []
+       preload_argv: []
+
+.. note::
+
+   Because the ``dask-worker`` command needs to accept keywords for both the
+   Worker and the Nanny (if a nanny is used) it has both a ``--preload`` and
+   ``--preload-nanny`` keyword.  All extra keywords (like ``--print-count``
+   above) will be sent to the workers rather than the nanny.  There is no way
+   to specify extra keywords to the nanny preload scripts on the command line.
+   We recommend the use of the more flexible configuration if this is
+   necessary.
+
 
 Worker Lifecycle Plugins
 ------------------------
@@ -73,4 +118,4 @@ and register that class with the scheduler to give to every worker using the
    Client.register_worker_plugin
 
 .. automethod:: Client.register_worker_plugin
-   :noindex:
\ No newline at end of file
+   :noindex:


Test patch:
diff --git a/dask/array/tests/test_creation.py b/dask/array/tests/test_creation.py
index 704f13e3d..ed27d227c 100644
--- a/dask/array/tests/test_creation.py
+++ b/dask/array/tests/test_creation.py
@@ -10,7 +10,7 @@ import dask
 import dask.array as da
 from dask.array.core import normalize_chunks
 from dask.array.utils import assert_eq, same_keys, AxisError
-from dask.array.numpy_compat import _numpy_117
+from dask.array.numpy_compat import _numpy_117, _numpy_118
 
 
 @pytest.mark.parametrize(
@@ -540,7 +540,7 @@ def test_repeat():
     x = np.random.random((10, 11, 13))
     d = da.from_array(x, chunks=(4, 5, 3))
 
-    repeats = [1, 2, 5]
+    repeats = [0, 1, 2, 5]
     axes = [-3, -2, -1, 0, 1, 2]
 
     for r in repeats:
@@ -719,6 +719,66 @@ def test_pad(shape, chunks, pad_width, mode, kwargs):
         assert_eq(np_r, da_r)
 
 
+@pytest.mark.parametrize("dtype", [np.uint8, np.int16, np.float32, bool])
+@pytest.mark.parametrize(
+    "pad_widths", [2, (2,), (2, 3), ((2, 3),), ((3, 1), (0, 0), (2, 0))]
+)
+@pytest.mark.parametrize(
+    "mode",
+    [
+        "constant",
+        "edge",
+        pytest.param(
+            "linear_ramp",
+            marks=pytest.mark.skipif(
+                not _numpy_118, reason="numpy changed pad behaviour"
+            ),
+        ),
+        "maximum",
+        pytest.param(
+            "mean",
+            marks=pytest.mark.skip(
+                reason="Bug dask changes the dtype to float: https://github.com/dask/dask/issues/5303"
+            ),
+        ),
+        "minimum",
+        pytest.param(
+            "reflect",
+            marks=pytest.mark.skip(
+                reason="Bug when pad_width is larger than dimension: https://github.com/dask/dask/issues/5303"
+            ),
+        ),
+        pytest.param(
+            "symmetric",
+            marks=pytest.mark.skip(
+                reason="Bug when pad_width is larger than dimension: https://github.com/dask/dask/issues/5303"
+            ),
+        ),
+        pytest.param(
+            "wrap",
+            marks=pytest.mark.skip(
+                reason="Bug when pad_width is larger than dimension: https://github.com/dask/dask/issues/5303"
+            ),
+        ),
+        pytest.param("median", marks=pytest.mark.skip(reason="Not implemented"),),
+        pytest.param(
+            "empty",
+            marks=pytest.mark.skip(
+                reason="Empty leads to undefined values, which may be different"
+            ),
+        ),
+    ],
+)
+def test_pad_3d_data(dtype, pad_widths, mode):
+    np_a = np.arange(2 * 3 * 4).reshape(2, 3, 4).astype(dtype)
+    da_a = da.from_array(np_a, chunks="auto")
+
+    np_r = np.pad(np_a, pad_widths, mode=mode)
+    da_r = da.pad(da_a, pad_widths, mode=mode)
+
+    assert_eq(np_r, da_r)
+
+
 @pytest.mark.parametrize("kwargs", [{}, {"scaler": 2}])
 def test_pad_udf(kwargs):
     def udf_pad(vector, pad_width, iaxis, kwargs):
diff --git a/dask/array/tests/test_optimization.py b/dask/array/tests/test_optimization.py
index 064cc5900..bef4722e4 100644
--- a/dask/array/tests/test_optimization.py
+++ b/dask/array/tests/test_optimization.py
@@ -363,7 +363,7 @@ def test_turn_off_fusion():
 
     a = y.__dask_optimize__(y.dask, y.__dask_keys__())
 
-    with dask.config.set(fuse_ave_width=0):
+    with dask.config.set({"optimization.fuse.ave-width": 0}):
         b = y.__dask_optimize__(y.dask, y.__dask_keys__())
 
     assert dask.get(a, y.__dask_keys__()) == dask.get(b, y.__dask_keys__())
diff --git a/dask/bag/tests/test_bag.py b/dask/bag/tests/test_bag.py
index 428a78a7e..c30aee60f 100644
--- a/dask/bag/tests/test_bag.py
+++ b/dask/bag/tests/test_bag.py
@@ -563,7 +563,7 @@ def test_rename_fused_keys_bag():
     assert outp.keys() == {"c"}
     assert outp["c"][1:] == ("a", 1)
 
-    with dask.config.set({"fuse_rename_keys": False}):
+    with dask.config.set({"optimization.fuse.rename-keys": False}):
         assert optimize(inp, ["c"]) == outp
 
     # By default, fused keys are renamed
diff --git a/dask/dataframe/io/tests/test_parquet.py b/dask/dataframe/io/tests/test_parquet.py
index aace56121..1973209fa 100644
--- a/dask/dataframe/io/tests/test_parquet.py
+++ b/dask/dataframe/io/tests/test_parquet.py
@@ -1,4 +1,5 @@
 import os
+import sys
 import warnings
 from distutils.version import LooseVersion
 
@@ -43,8 +44,12 @@ if pq and pa.__version__ < LooseVersion("0.13.1"):
     SKIP_PYARROW = True
     SKIP_PYARROW_REASON = "pyarrow >= 0.13.1 required for parquet"
 else:
-    SKIP_PYARROW = not pq
-    SKIP_PYARROW_REASON = "pyarrow not found"
+    if sys.platform == "win32" and pa.__version__ == LooseVersion("0.16.0"):
+        SKIP_PYARROW = True
+        SKIP_PYARROW_REASON = "https://github.com/dask/dask/issues/6093"
+    else:
+        SKIP_PYARROW = not pq
+        SKIP_PYARROW_REASON = "pyarrow not found"
 PYARROW_MARK = pytest.mark.skipif(SKIP_PYARROW, reason=SKIP_PYARROW_REASON)
 
 
@@ -667,6 +672,8 @@ def test_partition_on_cats(tmpdir, engine):
 @pytest.mark.parametrize("meta", [False, True])
 @pytest.mark.parametrize("stats", [False, True])
 def test_partition_on_cats_pyarrow(tmpdir, stats, meta):
+    check_pyarrow()
+
     tmp = str(tmpdir)
     d = pd.DataFrame(
         {
@@ -2358,3 +2365,27 @@ def test_filter_nonpartition_columns(
     df_read = ddf_read.compute()
     assert len(df_read) == len(df_read[df_read["time"] < 5])
     assert df_read["time"].max() < 5
+
+
+def test_pandas_metadata_nullable_pyarrow(tmpdir):
+
+    check_pyarrow()
+    if pa.__version__ < LooseVersion("0.16.0") or pd.__version__ < LooseVersion(
+        "1.0.0"
+    ):
+        pytest.skip("PyArrow>=0.16 and Pandas>=1.0.0 Required.")
+    tmpdir = str(tmpdir)
+
+    ddf1 = dd.from_pandas(
+        pd.DataFrame(
+            {
+                "A": pd.array([1, None, 2], dtype="Int64"),
+                "B": pd.array(["dog", "cat", None], dtype="str"),
+            }
+        ),
+        npartitions=1,
+    )
+    ddf1.to_parquet(tmpdir, engine="pyarrow")
+    ddf2 = dd.read_parquet(tmpdir, engine="pyarrow")
+
+    assert_eq(ddf1, ddf2, check_index=False)
diff --git a/dask/dataframe/tests/test_dataframe.py b/dask/dataframe/tests/test_dataframe.py
index 48d789b71..b6fb7bdd7 100644
--- a/dask/dataframe/tests/test_dataframe.py
+++ b/dask/dataframe/tests/test_dataframe.py
@@ -1062,6 +1062,8 @@ def test_len():
     assert len(d.a) == len(full.a)
     assert len(dd.from_pandas(pd.DataFrame(), npartitions=1)) == 0
     assert len(dd.from_pandas(pd.DataFrame(columns=[1, 2]), npartitions=1)) == 0
+    # Regression test for https://github.com/dask/dask/issues/6110
+    assert len(dd.from_pandas(pd.DataFrame(columns=["foo", "foo"]), npartitions=1)) == 0
 
 
 def test_size():
diff --git a/dask/dataframe/tests/test_optimize_dataframe.py b/dask/dataframe/tests/test_optimize_dataframe.py
index 8c9f1d701..56a0f006b 100644
--- a/dask/dataframe/tests/test_optimize_dataframe.py
+++ b/dask/dataframe/tests/test_optimize_dataframe.py
@@ -16,7 +16,7 @@ def test_fuse_ave_width():
 
     s = (df.x + 1) + (df.x + 2)
 
-    with dask.config.set(fuse_ave_width=4):
+    with dask.config.set({"optimization.fuse.ave-width": 4}):
         a = s.__dask_optimize__(s.dask, s.__dask_keys__())
 
     b = s.__dask_optimize__(s.dask, s.__dask_keys__())
diff --git a/dask/tests/test_config.py b/dask/tests/test_config.py
index cecd046ca..480e0df78 100644
--- a/dask/tests/test_config.py
+++ b/dask/tests/test_config.py
@@ -418,3 +418,11 @@ def test_core_file():
     assert "temporary-directory" in dask.config.config
     assert "dataframe" in dask.config.config
     assert "shuffle-compression" in dask.config.get("dataframe")
+
+
+def test_deprecations():
+    with pytest.warns(Warning) as info:
+        with dask.config.set(fuse_ave_width=123):
+            assert dask.config.get("optimization.fuse.ave-width") == 123
+
+    assert "optimization.fuse.ave-width" in str(info[0].message)
diff --git a/dask/tests/test_optimization.py b/dask/tests/test_optimization.py
index 99e32635f..8915aa730 100644
--- a/dask/tests/test_optimization.py
+++ b/dask/tests/test_optimization.py
@@ -4,6 +4,7 @@ from functools import partial
 
 import pytest
 
+import dask
 from dask.utils_test import add, inc
 from dask.core import get_dependencies
 from dask.local import get_sync
@@ -1297,3 +1298,49 @@ def test_dont_fuse_numpy_arrays():
     dsk = {"x": np.arange(5), "y": (inc, "x")}
 
     assert fuse(dsk, "y")[0] == dsk
+
+
+def test_fuse_config():
+    with dask.config.set({"optimization.fuse.active": False}):
+        d = {
+            "a": 1,
+            "b": (inc, "a"),
+        }
+        dependencies = {"b": ("a",)}
+        assert fuse(d, "b", dependencies=dependencies) == (d, dependencies)
+
+
+def test_fused_keys_max_length():  # generic fix for gh-5999
+    d = {
+        "u-looooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooong": (
+            inc,
+            "v-looooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooong",
+        ),
+        "v-looooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooong": (
+            inc,
+            "w-looooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooong",
+        ),
+        "w-looooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooong": (
+            inc,
+            "x-looooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooong",
+        ),
+        "x-looooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooong": (
+            inc,
+            "y-looooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooong",
+        ),
+        "y-looooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooong": (
+            inc,
+            "z-looooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooong",
+        ),
+        "z-looooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooooong": (
+            add,
+            "a",
+            "b",
+        ),
+        "a": 1,
+        "b": 2,
+    }
+
+    fused, deps = fuse(d, rename_keys=True)
+    for key in fused:
+        assert len(key) < 150