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.
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:

`_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.

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

pip install -e .[complete]
parse_log_pytest
requirements.txt
pytest
apt-get update apt-get install -y gcc
3.7
pytest --no-header -rA --tb=line --color=no -p no:cacheprovider -W ignore::DeprecationWarning
bokeh==2.4.3
certifi @ file:///croot/certifi_1671487769961/work/certifi
click==8.1.8
cloudpickle==2.2.1
-e git+https://github.com/dask/dask.git@007a8d7e9bce1f0d5b3677a553caf03ecb64d091#egg=dask
distributed==2.30.1
exceptiongroup==1.2.2
fsspec==2023.1.0
HeapDict==1.0.1
importlib-metadata==6.7.0
iniconfig==2.0.0
Jinja2==3.1.6
locket==1.0.0
MarkupSafe==2.1.5
msgpack==1.0.5
numpy==1.21.6
packaging==24.0
pandas==1.3.5
partd==1.4.1
Pillow==9.5.0
pluggy==1.2.0
psutil==7.0.0
pytest==7.4.4
python-dateutil==2.9.0.post0
pytz==2025.2
PyYAML==6.0.1
six==1.17.0
sortedcontainers==2.4.0
tblib==2.0.0
tomli==2.0.1
toolz==0.12.1
tornado==6.2
typing_extensions==4.7.1
zict==2.2.0
zipp==3.15.0

name: dask
channels:
  - defaults
  - https://repo.anaconda.com/pkgs/main
  - https://repo.anaconda.com/pkgs/r
  - conda-forge
dependencies:
  - _libgcc_mutex=0.1=main
  - _openmp_mutex=5.1=1_gnu
  - ca-certificates=2025.2.25=h06a4308_0
  - certifi=2022.12.7=py37h06a4308_0
  - ld_impl_linux-64=2.40=h12ee557_0
  - libffi=3.4.4=h6a678d5_1
  - libgcc-ng=11.2.0=h1234567_1
  - libgomp=11.2.0=h1234567_1
  - libstdcxx-ng=11.2.0=h1234567_1
  - ncurses=6.4=h6a678d5_0
  - openssl=1.1.1w=h7f8727e_0
  - pip=22.3.1=py37h06a4308_0
  - python=3.7.16=h7a1cb2a_0
  - readline=8.2=h5eee18b_0
  - setuptools=65.6.3=py37h06a4308_0
  - sqlite=3.45.3=h5eee18b_0
  - tk=8.6.14=h39e8969_0
  - wheel=0.38.4=py37h06a4308_0
  - xz=5.6.4=h5eee18b_1
  - zlib=1.2.13=h5eee18b_1
  - pip:
      - bokeh==2.4.3
      - click==8.1.8
      - cloudpickle==2.2.1
      - distributed==2.30.1
      - exceptiongroup==1.2.2
      - fsspec==2023.1.0
      - heapdict==1.0.1
      - importlib-metadata==6.7.0
      - iniconfig==2.0.0
      - jinja2==3.1.6
      - locket==1.0.0
      - markupsafe==2.1.5
      - msgpack==1.0.5
      - numpy==1.21.6
      - packaging==24.0
      - pandas==1.3.5
      - partd==1.4.1
      - pillow==9.5.0
      - pluggy==1.2.0
      - psutil==7.0.0
      - pytest==7.4.4
      - python-dateutil==2.9.0.post0
      - pytz==2025.2
      - pyyaml==6.0.1
      - six==1.17.0
      - sortedcontainers==2.4.0
      - tblib==2.0.0
      - tomli==2.0.1
      - toolz==0.12.1
      - tornado==6.2
      - typing-extensions==4.7.1
      - zict==2.2.0
      - zipp==3.15.0
prefix: /opt/conda/envs/dask

dask/array/tests/test_creation.py::test_repeat dask/array/tests/test_creation.py::test_pad_3d_data[constant-pad_widths3-uint8] dask/array/tests/test_creation.py::test_pad_3d_data[constant-pad_widths3-int16] dask/array/tests/test_creation.py::test_pad_3d_data[constant-pad_widths3-float32] dask/array/tests/test_creation.py::test_pad_3d_data[constant-pad_widths3-bool] dask/array/tests/test_creation.py::test_pad_3d_data[edge-pad_widths3-uint8] dask/array/tests/test_creation.py::test_pad_3d_data[edge-pad_widths3-int16] dask/array/tests/test_creation.py::test_pad_3d_data[edge-pad_widths3-float32] dask/array/tests/test_creation.py::test_pad_3d_data[edge-pad_widths3-bool] dask/array/tests/test_creation.py::test_pad_3d_data[linear_ramp-pad_widths3-uint8] dask/array/tests/test_creation.py::test_pad_3d_data[linear_ramp-pad_widths3-int16] dask/array/tests/test_creation.py::test_pad_3d_data[linear_ramp-pad_widths3-float32] dask/array/tests/test_creation.py::test_pad_3d_data[linear_ramp-pad_widths3-bool] dask/array/tests/test_creation.py::test_pad_3d_data[maximum-2-uint8] dask/array/tests/test_creation.py::test_pad_3d_data[maximum-2-int16] dask/array/tests/test_creation.py::test_pad_3d_data[maximum-2-float32] dask/array/tests/test_creation.py::test_pad_3d_data[maximum-2-bool] dask/array/tests/test_creation.py::test_pad_3d_data[maximum-pad_widths1-uint8] dask/array/tests/test_creation.py::test_pad_3d_data[maximum-pad_widths1-int16] dask/array/tests/test_creation.py::test_pad_3d_data[maximum-pad_widths1-float32] dask/array/tests/test_creation.py::test_pad_3d_data[maximum-pad_widths1-bool] dask/array/tests/test_creation.py::test_pad_3d_data[maximum-pad_widths2-uint8] dask/array/tests/test_creation.py::test_pad_3d_data[maximum-pad_widths2-int16] dask/array/tests/test_creation.py::test_pad_3d_data[maximum-pad_widths2-float32] dask/array/tests/test_creation.py::test_pad_3d_data[maximum-pad_widths2-bool] dask/array/tests/test_creation.py::test_pad_3d_data[maximum-pad_widths3-uint8] dask/array/tests/test_creation.py::test_pad_3d_data[maximum-pad_widths3-int16] dask/array/tests/test_creation.py::test_pad_3d_data[maximum-pad_widths3-float32] dask/array/tests/test_creation.py::test_pad_3d_data[maximum-pad_widths3-bool] dask/array/tests/test_creation.py::test_pad_3d_data[maximum-pad_widths4-uint8] dask/array/tests/test_creation.py::test_pad_3d_data[maximum-pad_widths4-int16] dask/array/tests/test_creation.py::test_pad_3d_data[maximum-pad_widths4-float32] dask/array/tests/test_creation.py::test_pad_3d_data[maximum-pad_widths4-bool] dask/array/tests/test_creation.py::test_pad_3d_data[minimum-2-uint8] dask/array/tests/test_creation.py::test_pad_3d_data[minimum-2-int16] dask/array/tests/test_creation.py::test_pad_3d_data[minimum-2-float32] dask/array/tests/test_creation.py::test_pad_3d_data[minimum-2-bool] dask/array/tests/test_creation.py::test_pad_3d_data[minimum-pad_widths1-uint8] dask/array/tests/test_creation.py::test_pad_3d_data[minimum-pad_widths1-int16] dask/array/tests/test_creation.py::test_pad_3d_data[minimum-pad_widths1-float32] dask/array/tests/test_creation.py::test_pad_3d_data[minimum-pad_widths1-bool] dask/array/tests/test_creation.py::test_pad_3d_data[minimum-pad_widths2-uint8] dask/array/tests/test_creation.py::test_pad_3d_data[minimum-pad_widths2-int16] dask/array/tests/test_creation.py::test_pad_3d_data[minimum-pad_widths2-float32] dask/array/tests/test_creation.py::test_pad_3d_data[minimum-pad_widths2-bool] dask/array/tests/test_creation.py::test_pad_3d_data[minimum-pad_widths3-uint8] dask/array/tests/test_creation.py::test_pad_3d_data[minimum-pad_widths3-int16] dask/array/tests/test_creation.py::test_pad_3d_data[minimum-pad_widths3-float32] dask/array/tests/test_creation.py::test_pad_3d_data[minimum-pad_widths3-bool] dask/array/tests/test_creation.py::test_pad_3d_data[minimum-pad_widths4-uint8] dask/array/tests/test_creation.py::test_pad_3d_data[minimum-pad_widths4-int16] dask/array/tests/test_creation.py::test_pad_3d_data[minimum-pad_widths4-float32] dask/array/tests/test_creation.py::test_pad_3d_data[minimum-pad_widths4-bool] dask/bag/tests/test_bag.py::test_rename_fused_keys_bag dask/dataframe/tests/test_dataframe.py::test_len dask/tests/test_config.py::test_deprecations dask/tests/test_optimization.py::test_fuse_config dask/tests/test_optimization.py::test_fused_keys_max_length
dask/dataframe/tests/test_dataframe.py::test_timezone_freq[1] dask/dataframe/tests/test_dataframe.py::test_describe[None-None-None-subset1] dask/dataframe/tests/test_dataframe.py::test_describe[None-None-None-subset3] dask/dataframe/tests/test_dataframe.py::test_describe[None-None-None-subset4] dask/dataframe/tests/test_dataframe.py::test_describe[all-None-None-None] dask/dataframe/tests/test_dataframe.py::test_describe[include6-None-percentiles6-None] dask/dataframe/tests/test_dataframe.py::test_describe[include7-None-None-None] dask/dataframe/tests/test_dataframe.py::test_describe[include8-None-percentiles8-None] dask/dataframe/tests/test_dataframe.py::test_describe[None-exclude9-None-None] dask/dataframe/tests/test_dataframe.py::test_describe[include10-None-None-None] dask/dataframe/tests/test_dataframe.py::test_embarrassingly_parallel_operations dask/dataframe/tests/test_dataframe.py::test_to_timestamp dask/dataframe/tests/test_dataframe.py::test_apply_warns dask/dataframe/tests/test_dataframe.py::test_astype_categoricals dask/dataframe/tests/test_dataframe.py::test_astype_categoricals_known dask/dataframe/tests/test_dataframe.py::test_idxmaxmin[idx2-True] dask/dataframe/tests/test_dataframe.py::test_idxmaxmin[idx2-False] dask/dataframe/tests/test_dataframe.py::test_first_and_last[first] dask/dataframe/tests/test_dataframe.py::test_datetime_loc_open_slicing dask/tests/test_config.py::test_collect_yaml_permission_errors[directory] dask/tests/test_config.py::test_collect_yaml_permission_errors[file]
dask/array/tests/test_creation.py::test_arr_like[i4-shape0-chunks0-tuple-tuple-empty_like] dask/array/tests/test_creation.py::test_arr_like[i4-shape0-chunks0-tuple-tuple-empty] dask/array/tests/test_creation.py::test_arr_like[i4-shape0-chunks0-tuple-tuple-ones_like] dask/array/tests/test_creation.py::test_arr_like[i4-shape0-chunks0-tuple-tuple-ones] dask/array/tests/test_creation.py::test_arr_like[i4-shape0-chunks0-tuple-tuple-zeros_like] dask/array/tests/test_creation.py::test_arr_like[i4-shape0-chunks0-tuple-tuple-zeros] dask/array/tests/test_creation.py::test_arr_like[i4-shape0-chunks0-tuple-tuple-full_like] dask/array/tests/test_creation.py::test_arr_like[i4-shape0-chunks0-tuple-tuple-full] dask/array/tests/test_creation.py::test_arr_like[i4-shape0-chunks0-tuple-list-empty_like] dask/array/tests/test_creation.py::test_arr_like[i4-shape0-chunks0-tuple-list-empty] dask/array/tests/test_creation.py::test_arr_like[i4-shape0-chunks0-tuple-list-ones_like] dask/array/tests/test_creation.py::test_arr_like[i4-shape0-chunks0-tuple-list-ones] dask/array/tests/test_creation.py::test_arr_like[i4-shape0-chunks0-tuple-list-zeros_like] dask/array/tests/test_creation.py::test_arr_like[i4-shape0-chunks0-tuple-list-zeros] dask/array/tests/test_creation.py::test_arr_like[i4-shape0-chunks0-tuple-list-full_like] dask/array/tests/test_creation.py::test_arr_like[i4-shape0-chunks0-tuple-list-full] dask/array/tests/test_creation.py::test_arr_like[i4-shape0-chunks0-tuple-asarray-empty_like] dask/array/tests/test_creation.py::test_arr_like[i4-shape0-chunks0-tuple-asarray-empty] dask/array/tests/test_creation.py::test_arr_like[i4-shape0-chunks0-tuple-asarray-ones_like] dask/array/tests/test_creation.py::test_arr_like[i4-shape0-chunks0-tuple-asarray-ones] dask/array/tests/test_creation.py::test_arr_like[i4-shape0-chunks0-tuple-asarray-zeros_like] dask/array/tests/test_creation.py::test_arr_like[i4-shape0-chunks0-tuple-asarray-zeros] dask/array/tests/test_creation.py::test_arr_like[i4-shape0-chunks0-tuple-asarray-full_like] dask/array/tests/test_creation.py::test_arr_like[i4-shape0-chunks0-tuple-asarray-full] dask/array/tests/test_creation.py::test_arr_like[i4-shape0-chunks0-list-tuple-empty_like] dask/array/tests/test_creation.py::test_arr_like[i4-shape0-chunks0-list-tuple-empty] dask/array/tests/test_creation.py::test_arr_like[i4-shape0-chunks0-list-tuple-ones_like] dask/array/tests/test_creation.py::test_arr_like[i4-shape0-chunks0-list-tuple-ones] dask/array/tests/test_creation.py::test_arr_like[i4-shape0-chunks0-list-tuple-zeros_like] dask/array/tests/test_creation.py::test_arr_like[i4-shape0-chunks0-list-tuple-zeros] dask/array/tests/test_creation.py::test_arr_like[i4-shape0-chunks0-list-tuple-full_like] dask/array/tests/test_creation.py::test_arr_like[i4-shape0-chunks0-list-tuple-full] dask/array/tests/test_creation.py::test_arr_like[i4-shape0-chunks0-list-list-empty_like] dask/array/tests/test_creation.py::test_arr_like[i4-shape0-chunks0-list-list-empty] dask/array/tests/test_creation.py::test_arr_like[i4-shape0-chunks0-list-list-ones_like] dask/array/tests/test_creation.py::test_arr_like[i4-shape0-chunks0-list-list-ones] dask/array/tests/test_creation.py::test_arr_like[i4-shape0-chunks0-list-list-zeros_like] dask/array/tests/test_creation.py::test_arr_like[i4-shape0-chunks0-list-list-zeros] dask/array/tests/test_creation.py::test_arr_like[i4-shape0-chunks0-list-list-full_like] dask/array/tests/test_creation.py::test_arr_like[i4-shape0-chunks0-list-list-full] dask/array/tests/test_creation.py::test_arr_like[i4-shape0-chunks0-list-asarray-empty_like] dask/array/tests/test_creation.py::test_arr_like[i4-shape0-chunks0-list-asarray-empty] dask/array/tests/test_creation.py::test_arr_like[i4-shape0-chunks0-list-asarray-ones_like] dask/array/tests/test_creation.py::test_arr_like[i4-shape0-chunks0-list-asarray-ones] dask/array/tests/test_creation.py::test_arr_like[i4-shape0-chunks0-list-asarray-zeros_like] dask/array/tests/test_creation.py::test_arr_like[i4-shape0-chunks0-list-asarray-zeros] dask/array/tests/test_creation.py::test_arr_like[i4-shape0-chunks0-list-asarray-full_like] dask/array/tests/test_creation.py::test_arr_like[i4-shape0-chunks0-list-asarray-full] dask/array/tests/test_creation.py::test_arr_like[i4-shape0-chunks0-asarray-tuple-empty_like] dask/array/tests/test_creation.py::test_arr_like[i4-shape0-chunks0-asarray-tuple-empty] dask/array/tests/test_creation.py::test_arr_like[i4-shape0-chunks0-asarray-tuple-ones_like] dask/array/tests/test_creation.py::test_arr_like[i4-shape0-chunks0-asarray-tuple-ones] dask/array/tests/test_creation.py::test_arr_like[i4-shape0-chunks0-asarray-tuple-zeros_like] dask/array/tests/test_creation.py::test_arr_like[i4-shape0-chunks0-asarray-tuple-zeros] dask/array/tests/test_creation.py::test_arr_like[i4-shape0-chunks0-asarray-tuple-full_like] dask/array/tests/test_creation.py::test_arr_like[i4-shape0-chunks0-asarray-tuple-full] dask/array/tests/test_creation.py::test_arr_like[i4-shape0-chunks0-asarray-list-empty_like] dask/array/tests/test_creation.py::test_arr_like[i4-shape0-chunks0-asarray-list-empty] dask/array/tests/test_creation.py::test_arr_like[i4-shape0-chunks0-asarray-list-ones_like] dask/array/tests/test_creation.py::test_arr_like[i4-shape0-chunks0-asarray-list-ones] dask/array/tests/test_creation.py::test_arr_like[i4-shape0-chunks0-asarray-list-zeros_like] dask/array/tests/test_creation.py::test_arr_like[i4-shape0-chunks0-asarray-list-zeros] dask/array/tests/test_creation.py::test_arr_like[i4-shape0-chunks0-asarray-list-full_like] dask/array/tests/test_creation.py::test_arr_like[i4-shape0-chunks0-asarray-list-full] dask/array/tests/test_creation.py::test_arr_like[i4-shape0-chunks0-asarray-asarray-empty_like] dask/array/tests/test_creation.py::test_arr_like[i4-shape0-chunks0-asarray-asarray-empty] dask/array/tests/test_creation.py::test_arr_like[i4-shape0-chunks0-asarray-asarray-ones_like] dask/array/tests/test_creation.py::test_arr_like[i4-shape0-chunks0-asarray-asarray-ones] dask/array/tests/test_creation.py::test_arr_like[i4-shape0-chunks0-asarray-asarray-zeros_like] dask/array/tests/test_creation.py::test_arr_like[i4-shape0-chunks0-asarray-asarray-zeros] dask/array/tests/test_creation.py::test_arr_like[i4-shape0-chunks0-asarray-asarray-full_like] dask/array/tests/test_creation.py::test_arr_like[i4-shape0-chunks0-asarray-asarray-full] dask/array/tests/test_creation.py::test_linspace[True] dask/array/tests/test_creation.py::test_linspace[False] dask/array/tests/test_creation.py::test_arange dask/array/tests/test_creation.py::test_arange_dtypes[0-1-1-None] dask/array/tests/test_creation.py::test_arange_dtypes[1.5-2-1-None] dask/array/tests/test_creation.py::test_arange_dtypes[1-2.5-1-None] dask/array/tests/test_creation.py::test_arange_dtypes[1-2-0.5-None] dask/array/tests/test_creation.py::test_arange_dtypes[start4-stop4-step4-None] dask/array/tests/test_creation.py::test_arange_dtypes[start5-stop5-step5-None] dask/array/tests/test_creation.py::test_arange_dtypes[start6-stop6-step6-None] dask/array/tests/test_creation.py::test_arange_dtypes[start7-stop7-step7-None] dask/array/tests/test_creation.py::test_arange_dtypes[start8-stop8-step8-uint32] dask/array/tests/test_creation.py::test_arange_dtypes[start9-stop9-step9-uint64] dask/array/tests/test_creation.py::test_arange_float_step dask/array/tests/test_creation.py::test_indices_wrong_chunks dask/array/tests/test_creation.py::test_indices_dimensions_chunks dask/array/tests/test_creation.py::test_empty_indicies dask/array/tests/test_creation.py::test_indicies dask/array/tests/test_creation.py::test_meshgrid[False-ij-shapes0-chunks0] dask/array/tests/test_creation.py::test_meshgrid[False-ij-shapes1-chunks1] dask/array/tests/test_creation.py::test_meshgrid[False-ij-shapes2-chunks2] dask/array/tests/test_creation.py::test_meshgrid[False-ij-shapes3-chunks3] dask/array/tests/test_creation.py::test_meshgrid[False-ij-shapes4-chunks4] dask/array/tests/test_creation.py::test_meshgrid[False-ij-shapes5-chunks5] dask/array/tests/test_creation.py::test_meshgrid[False-xy-shapes0-chunks0] dask/array/tests/test_creation.py::test_meshgrid[False-xy-shapes1-chunks1] dask/array/tests/test_creation.py::test_meshgrid[False-xy-shapes2-chunks2] dask/array/tests/test_creation.py::test_meshgrid[False-xy-shapes3-chunks3] dask/array/tests/test_creation.py::test_meshgrid[False-xy-shapes4-chunks4] dask/array/tests/test_creation.py::test_meshgrid[False-xy-shapes5-chunks5] dask/array/tests/test_creation.py::test_meshgrid[True-ij-shapes0-chunks0] dask/array/tests/test_creation.py::test_meshgrid[True-ij-shapes1-chunks1] dask/array/tests/test_creation.py::test_meshgrid[True-ij-shapes2-chunks2] dask/array/tests/test_creation.py::test_meshgrid[True-ij-shapes3-chunks3] dask/array/tests/test_creation.py::test_meshgrid[True-ij-shapes4-chunks4] dask/array/tests/test_creation.py::test_meshgrid[True-ij-shapes5-chunks5] dask/array/tests/test_creation.py::test_meshgrid[True-xy-shapes0-chunks0] dask/array/tests/test_creation.py::test_meshgrid[True-xy-shapes1-chunks1] dask/array/tests/test_creation.py::test_meshgrid[True-xy-shapes2-chunks2] dask/array/tests/test_creation.py::test_meshgrid[True-xy-shapes3-chunks3] dask/array/tests/test_creation.py::test_meshgrid[True-xy-shapes4-chunks4] dask/array/tests/test_creation.py::test_meshgrid[True-xy-shapes5-chunks5] dask/array/tests/test_creation.py::test_meshgrid_inputcoercion dask/array/tests/test_creation.py::test_tril_triu dask/array/tests/test_creation.py::test_tril_triu_errors dask/array/tests/test_creation.py::test_tril_triu_non_square_arrays dask/array/tests/test_creation.py::test_eye dask/array/tests/test_creation.py::test_diag dask/array/tests/test_creation.py::test_diagonal dask/array/tests/test_creation.py::test_fromfunction[<lambda>-kwargs0-None] dask/array/tests/test_creation.py::test_fromfunction[<lambda>-kwargs0-f8] dask/array/tests/test_creation.py::test_fromfunction[<lambda>-kwargs0-i8] dask/array/tests/test_creation.py::test_fromfunction[<lambda>-kwargs1-None] dask/array/tests/test_creation.py::test_fromfunction[<lambda>-kwargs1-f8] dask/array/tests/test_creation.py::test_fromfunction[<lambda>-kwargs1-i8] dask/array/tests/test_creation.py::test_fromfunction[<lambda>-kwargs2-None] dask/array/tests/test_creation.py::test_fromfunction[<lambda>-kwargs2-f8] dask/array/tests/test_creation.py::test_fromfunction[<lambda>-kwargs2-i8] dask/array/tests/test_creation.py::test_tile_basic[2] dask/array/tests/test_creation.py::test_tile_basic[reps1] dask/array/tests/test_creation.py::test_tile_basic[reps2] dask/array/tests/test_creation.py::test_tile_basic[reps3] dask/array/tests/test_creation.py::test_tile_basic[reps4] dask/array/tests/test_creation.py::test_tile_chunks[0-shape0-chunks0] dask/array/tests/test_creation.py::test_tile_chunks[0-shape1-chunks1] dask/array/tests/test_creation.py::test_tile_chunks[1-shape0-chunks0] dask/array/tests/test_creation.py::test_tile_chunks[1-shape1-chunks1] dask/array/tests/test_creation.py::test_tile_chunks[2-shape0-chunks0] dask/array/tests/test_creation.py::test_tile_chunks[2-shape1-chunks1] dask/array/tests/test_creation.py::test_tile_chunks[3-shape0-chunks0] dask/array/tests/test_creation.py::test_tile_chunks[3-shape1-chunks1] dask/array/tests/test_creation.py::test_tile_chunks[5-shape0-chunks0] dask/array/tests/test_creation.py::test_tile_chunks[5-shape1-chunks1] dask/array/tests/test_creation.py::test_tile_chunks[reps5-shape0-chunks0] dask/array/tests/test_creation.py::test_tile_chunks[reps5-shape1-chunks1] dask/array/tests/test_creation.py::test_tile_chunks[reps6-shape0-chunks0] dask/array/tests/test_creation.py::test_tile_chunks[reps6-shape1-chunks1] dask/array/tests/test_creation.py::test_tile_neg_reps[-1-shape0-chunks0] dask/array/tests/test_creation.py::test_tile_neg_reps[-1-shape1-chunks1] dask/array/tests/test_creation.py::test_tile_neg_reps[-5-shape0-chunks0] dask/array/tests/test_creation.py::test_tile_neg_reps[-5-shape1-chunks1] dask/array/tests/test_creation.py::test_tile_zero_reps[0-shape0-chunks0] dask/array/tests/test_creation.py::test_tile_zero_reps[0-shape1-chunks1] dask/array/tests/test_creation.py::test_tile_zero_reps[reps1-shape0-chunks0] dask/array/tests/test_creation.py::test_tile_zero_reps[reps1-shape1-chunks1] dask/array/tests/test_creation.py::test_tile_zero_reps[reps2-shape0-chunks0] dask/array/tests/test_creation.py::test_tile_zero_reps[reps2-shape1-chunks1] dask/array/tests/test_creation.py::test_tile_zero_reps[reps3-shape0-chunks0] dask/array/tests/test_creation.py::test_tile_zero_reps[reps3-shape1-chunks1] dask/array/tests/test_creation.py::test_tile_empty_array[2-shape0-chunks0] dask/array/tests/test_creation.py::test_tile_empty_array[2-shape1-chunks1] dask/array/tests/test_creation.py::test_tile_empty_array[reps1-shape0-chunks0] dask/array/tests/test_creation.py::test_tile_empty_array[reps1-shape1-chunks1] dask/array/tests/test_creation.py::test_tile_np_kroncompare_examples[reps0-shape0] dask/array/tests/test_creation.py::test_tile_np_kroncompare_examples[reps0-shape1] dask/array/tests/test_creation.py::test_tile_np_kroncompare_examples[reps0-shape2] dask/array/tests/test_creation.py::test_tile_np_kroncompare_examples[reps0-shape3] dask/array/tests/test_creation.py::test_tile_np_kroncompare_examples[reps0-shape4] dask/array/tests/test_creation.py::test_tile_np_kroncompare_examples[reps0-shape5] dask/array/tests/test_creation.py::test_tile_np_kroncompare_examples[reps1-shape0] dask/array/tests/test_creation.py::test_tile_np_kroncompare_examples[reps1-shape1] dask/array/tests/test_creation.py::test_tile_np_kroncompare_examples[reps1-shape2] dask/array/tests/test_creation.py::test_tile_np_kroncompare_examples[reps1-shape3] dask/array/tests/test_creation.py::test_tile_np_kroncompare_examples[reps1-shape4] dask/array/tests/test_creation.py::test_tile_np_kroncompare_examples[reps1-shape5] dask/array/tests/test_creation.py::test_tile_np_kroncompare_examples[reps2-shape0] dask/array/tests/test_creation.py::test_tile_np_kroncompare_examples[reps2-shape1] dask/array/tests/test_creation.py::test_tile_np_kroncompare_examples[reps2-shape2] dask/array/tests/test_creation.py::test_tile_np_kroncompare_examples[reps2-shape3] dask/array/tests/test_creation.py::test_tile_np_kroncompare_examples[reps2-shape4] dask/array/tests/test_creation.py::test_tile_np_kroncompare_examples[reps2-shape5] dask/array/tests/test_creation.py::test_tile_np_kroncompare_examples[reps3-shape0] dask/array/tests/test_creation.py::test_tile_np_kroncompare_examples[reps3-shape1] dask/array/tests/test_creation.py::test_tile_np_kroncompare_examples[reps3-shape2] dask/array/tests/test_creation.py::test_tile_np_kroncompare_examples[reps3-shape3] dask/array/tests/test_creation.py::test_tile_np_kroncompare_examples[reps3-shape4] dask/array/tests/test_creation.py::test_tile_np_kroncompare_examples[reps3-shape5] dask/array/tests/test_creation.py::test_tile_np_kroncompare_examples[reps4-shape0] dask/array/tests/test_creation.py::test_tile_np_kroncompare_examples[reps4-shape1] dask/array/tests/test_creation.py::test_tile_np_kroncompare_examples[reps4-shape2] dask/array/tests/test_creation.py::test_tile_np_kroncompare_examples[reps4-shape3] dask/array/tests/test_creation.py::test_tile_np_kroncompare_examples[reps4-shape4] dask/array/tests/test_creation.py::test_tile_np_kroncompare_examples[reps4-shape5] dask/array/tests/test_creation.py::test_tile_np_kroncompare_examples[reps5-shape0] dask/array/tests/test_creation.py::test_tile_np_kroncompare_examples[reps5-shape1] dask/array/tests/test_creation.py::test_tile_np_kroncompare_examples[reps5-shape2] dask/array/tests/test_creation.py::test_tile_np_kroncompare_examples[reps5-shape3] dask/array/tests/test_creation.py::test_tile_np_kroncompare_examples[reps5-shape4] dask/array/tests/test_creation.py::test_tile_np_kroncompare_examples[reps5-shape5] dask/array/tests/test_creation.py::test_pad_0_width[shape0-chunks0-0-constant-kwargs0] dask/array/tests/test_creation.py::test_pad_0_width[shape1-chunks1-0-edge-kwargs1] dask/array/tests/test_creation.py::test_pad_0_width[shape2-chunks2-0-linear_ramp-kwargs2] dask/array/tests/test_creation.py::test_pad_0_width[shape3-chunks3-0-reflect-kwargs3] dask/array/tests/test_creation.py::test_pad_0_width[shape4-chunks4-0-symmetric-kwargs4] dask/array/tests/test_creation.py::test_pad_0_width[shape5-chunks5-0-wrap-kwargs5] dask/array/tests/test_creation.py::test_pad_0_width[shape6-chunks6-0-empty-kwargs6] dask/array/tests/test_creation.py::test_pad[shape0-chunks0-1-constant-kwargs0] dask/array/tests/test_creation.py::test_pad[shape1-chunks1-2-constant-kwargs1] dask/array/tests/test_creation.py::test_pad[shape2-chunks2-pad_width2-constant-kwargs2] dask/array/tests/test_creation.py::test_pad[shape3-chunks3-pad_width3-constant-kwargs3] dask/array/tests/test_creation.py::test_pad[shape4-chunks4-3-edge-kwargs4] dask/array/tests/test_creation.py::test_pad[shape5-chunks5-3-linear_ramp-kwargs5] dask/array/tests/test_creation.py::test_pad[shape6-chunks6-3-linear_ramp-kwargs6] dask/array/tests/test_creation.py::test_pad[shape7-chunks7-pad_width7-linear_ramp-kwargs7] dask/array/tests/test_creation.py::test_pad[shape8-chunks8-pad_width8-reflect-kwargs8] dask/array/tests/test_creation.py::test_pad[shape9-chunks9-pad_width9-symmetric-kwargs9] dask/array/tests/test_creation.py::test_pad[shape10-chunks10-pad_width10-wrap-kwargs10] dask/array/tests/test_creation.py::test_pad[shape11-chunks11-pad_width11-maximum-kwargs11] dask/array/tests/test_creation.py::test_pad[shape12-chunks12-pad_width12-mean-kwargs12] dask/array/tests/test_creation.py::test_pad[shape13-chunks13-pad_width13-minimum-kwargs13] dask/array/tests/test_creation.py::test_pad[shape14-chunks14-1-empty-kwargs14] dask/array/tests/test_creation.py::test_pad_3d_data[constant-2-uint8] dask/array/tests/test_creation.py::test_pad_3d_data[constant-2-int16] dask/array/tests/test_creation.py::test_pad_3d_data[constant-2-float32] dask/array/tests/test_creation.py::test_pad_3d_data[constant-2-bool] dask/array/tests/test_creation.py::test_pad_3d_data[constant-pad_widths1-uint8] dask/array/tests/test_creation.py::test_pad_3d_data[constant-pad_widths1-int16] dask/array/tests/test_creation.py::test_pad_3d_data[constant-pad_widths1-float32] dask/array/tests/test_creation.py::test_pad_3d_data[constant-pad_widths1-bool] dask/array/tests/test_creation.py::test_pad_3d_data[constant-pad_widths2-uint8] dask/array/tests/test_creation.py::test_pad_3d_data[constant-pad_widths2-int16] dask/array/tests/test_creation.py::test_pad_3d_data[constant-pad_widths2-float32] dask/array/tests/test_creation.py::test_pad_3d_data[constant-pad_widths2-bool] dask/array/tests/test_creation.py::test_pad_3d_data[constant-pad_widths4-uint8] dask/array/tests/test_creation.py::test_pad_3d_data[constant-pad_widths4-int16] dask/array/tests/test_creation.py::test_pad_3d_data[constant-pad_widths4-float32] dask/array/tests/test_creation.py::test_pad_3d_data[constant-pad_widths4-bool] dask/array/tests/test_creation.py::test_pad_3d_data[edge-2-uint8] dask/array/tests/test_creation.py::test_pad_3d_data[edge-2-int16] dask/array/tests/test_creation.py::test_pad_3d_data[edge-2-float32] dask/array/tests/test_creation.py::test_pad_3d_data[edge-2-bool] dask/array/tests/test_creation.py::test_pad_3d_data[edge-pad_widths1-uint8] dask/array/tests/test_creation.py::test_pad_3d_data[edge-pad_widths1-int16] dask/array/tests/test_creation.py::test_pad_3d_data[edge-pad_widths1-float32] dask/array/tests/test_creation.py::test_pad_3d_data[edge-pad_widths1-bool] dask/array/tests/test_creation.py::test_pad_3d_data[edge-pad_widths2-uint8] dask/array/tests/test_creation.py::test_pad_3d_data[edge-pad_widths2-int16] dask/array/tests/test_creation.py::test_pad_3d_data[edge-pad_widths2-float32] dask/array/tests/test_creation.py::test_pad_3d_data[edge-pad_widths2-bool] dask/array/tests/test_creation.py::test_pad_3d_data[edge-pad_widths4-uint8] dask/array/tests/test_creation.py::test_pad_3d_data[edge-pad_widths4-int16] dask/array/tests/test_creation.py::test_pad_3d_data[edge-pad_widths4-float32] dask/array/tests/test_creation.py::test_pad_3d_data[edge-pad_widths4-bool] dask/array/tests/test_creation.py::test_pad_3d_data[linear_ramp-2-uint8] dask/array/tests/test_creation.py::test_pad_3d_data[linear_ramp-2-int16] dask/array/tests/test_creation.py::test_pad_3d_data[linear_ramp-2-float32] dask/array/tests/test_creation.py::test_pad_3d_data[linear_ramp-2-bool] dask/array/tests/test_creation.py::test_pad_3d_data[linear_ramp-pad_widths1-uint8] dask/array/tests/test_creation.py::test_pad_3d_data[linear_ramp-pad_widths1-int16] dask/array/tests/test_creation.py::test_pad_3d_data[linear_ramp-pad_widths1-float32] dask/array/tests/test_creation.py::test_pad_3d_data[linear_ramp-pad_widths1-bool] dask/array/tests/test_creation.py::test_pad_3d_data[linear_ramp-pad_widths2-uint8] dask/array/tests/test_creation.py::test_pad_3d_data[linear_ramp-pad_widths2-int16] dask/array/tests/test_creation.py::test_pad_3d_data[linear_ramp-pad_widths2-float32] dask/array/tests/test_creation.py::test_pad_3d_data[linear_ramp-pad_widths2-bool] dask/array/tests/test_creation.py::test_pad_3d_data[linear_ramp-pad_widths4-uint8] dask/array/tests/test_creation.py::test_pad_3d_data[linear_ramp-pad_widths4-int16] dask/array/tests/test_creation.py::test_pad_3d_data[linear_ramp-pad_widths4-float32] dask/array/tests/test_creation.py::test_pad_3d_data[linear_ramp-pad_widths4-bool] dask/array/tests/test_creation.py::test_pad_udf[kwargs0] dask/array/tests/test_creation.py::test_pad_udf[kwargs1] dask/array/tests/test_creation.py::test_auto_chunks dask/array/tests/test_optimization.py::test_fuse_getitem dask/array/tests/test_optimization.py::test_fuse_getitem_lock dask/array/tests/test_optimization.py::test_optimize_with_getitem_fusion dask/array/tests/test_optimization.py::test_optimize_slicing dask/array/tests/test_optimization.py::test_fuse_slice dask/array/tests/test_optimization.py::test_fuse_slice_with_lists dask/array/tests/test_optimization.py::test_nonfusible_fancy_indexing dask/array/tests/test_optimization.py::test_hard_fuse_slice_cases dask/array/tests/test_optimization.py::test_dont_fuse_numpy_arrays dask/array/tests/test_optimization.py::test_minimize_data_transfer dask/array/tests/test_optimization.py::test_fuse_slices_with_alias dask/array/tests/test_optimization.py::test_dont_fuse_fancy_indexing_in_getter_nofancy dask/array/tests/test_optimization.py::test_fuse_getter_with_asarray[10] dask/array/tests/test_optimization.py::test_fuse_getter_with_asarray[5] dask/array/tests/test_optimization.py::test_fuse_getter_with_asarray[3] dask/array/tests/test_optimization.py::test_remove_no_op_slices_if_get_is_not_getter_or_getter_nofancy[getter-False] dask/array/tests/test_optimization.py::test_remove_no_op_slices_if_get_is_not_getter_or_getter_nofancy[getter_nofancy-False] dask/array/tests/test_optimization.py::test_remove_no_op_slices_if_get_is_not_getter_or_getter_nofancy[getitem-True] dask/array/tests/test_optimization.py::test_gh3937 dask/array/tests/test_optimization.py::test_double_dependencies dask/array/tests/test_optimization.py::test_fuse_roots dask/bag/tests/test_bag.py::test_Bag dask/bag/tests/test_bag.py::test_keys dask/bag/tests/test_bag.py::test_bag_map dask/bag/tests/test_bag.py::test_map_method dask/bag/tests/test_bag.py::test_starmap dask/bag/tests/test_bag.py::test_filter dask/bag/tests/test_bag.py::test_remove dask/bag/tests/test_bag.py::test_iter dask/bag/tests/test_bag.py::test_repr[str] dask/bag/tests/test_bag.py::test_repr[repr] dask/bag/tests/test_bag.py::test_pluck dask/bag/tests/test_bag.py::test_pluck_with_default dask/bag/tests/test_bag.py::test_unzip dask/bag/tests/test_bag.py::test_fold dask/bag/tests/test_bag.py::test_fold_bag dask/bag/tests/test_bag.py::test_distinct dask/bag/tests/test_bag.py::test_distinct_with_key dask/bag/tests/test_bag.py::test_frequencies dask/bag/tests/test_bag.py::test_frequencies_sorted dask/bag/tests/test_bag.py::test_topk dask/bag/tests/test_bag.py::test_topk_with_non_callable_key[1] dask/bag/tests/test_bag.py::test_topk_with_non_callable_key[2] dask/bag/tests/test_bag.py::test_topk_with_multiarg_lambda dask/bag/tests/test_bag.py::test_lambdas dask/bag/tests/test_bag.py::test_reductions dask/bag/tests/test_bag.py::test_reduction_names dask/bag/tests/test_bag.py::test_tree_reductions dask/bag/tests/test_bag.py::test_aggregation[1] dask/bag/tests/test_bag.py::test_aggregation[3] dask/bag/tests/test_bag.py::test_aggregation[4] dask/bag/tests/test_bag.py::test_non_splittable_reductions[1] dask/bag/tests/test_bag.py::test_non_splittable_reductions[10] dask/bag/tests/test_bag.py::test_std dask/bag/tests/test_bag.py::test_var dask/bag/tests/test_bag.py::test_join[identity] dask/bag/tests/test_bag.py::test_join[delayed] dask/bag/tests/test_bag.py::test_join[<lambda>] dask/bag/tests/test_bag.py::test_foldby dask/bag/tests/test_bag.py::test_foldby_tree_reduction dask/bag/tests/test_bag.py::test_map_partitions dask/bag/tests/test_bag.py::test_map_partitions_args_kwargs dask/bag/tests/test_bag.py::test_random_sample_size dask/bag/tests/test_bag.py::test_random_sample_prob_range dask/bag/tests/test_bag.py::test_random_sample_repeated_computation dask/bag/tests/test_bag.py::test_random_sample_different_definitions dask/bag/tests/test_bag.py::test_random_sample_random_state dask/bag/tests/test_bag.py::test_lazify_task dask/bag/tests/test_bag.py::test_lazify dask/bag/tests/test_bag.py::test_inline_singleton_lists dask/bag/tests/test_bag.py::test_take dask/bag/tests/test_bag.py::test_take_npartitions dask/bag/tests/test_bag.py::test_take_npartitions_warn dask/bag/tests/test_bag.py::test_map_is_lazy dask/bag/tests/test_bag.py::test_can_use_dict_to_make_concrete dask/bag/tests/test_bag.py::test_read_text dask/bag/tests/test_bag.py::test_read_text_large dask/bag/tests/test_bag.py::test_read_text_encoding dask/bag/tests/test_bag.py::test_read_text_large_gzip dask/bag/tests/test_bag.py::test_from_sequence dask/bag/tests/test_bag.py::test_from_long_sequence dask/bag/tests/test_bag.py::test_from_empty_sequence dask/bag/tests/test_bag.py::test_product dask/bag/tests/test_bag.py::test_partition_collect dask/bag/tests/test_bag.py::test_groupby dask/bag/tests/test_bag.py::test_groupby_with_indexer dask/bag/tests/test_bag.py::test_groupby_with_npartitions_changed dask/bag/tests/test_bag.py::test_concat dask/bag/tests/test_bag.py::test_flatten dask/bag/tests/test_bag.py::test_concat_after_map dask/bag/tests/test_bag.py::test_args dask/bag/tests/test_bag.py::test_to_dataframe dask/bag/tests/test_bag.py::test_to_textfiles[gz-GzipFile] dask/bag/tests/test_bag.py::test_to_textfiles[bz2-BZ2File] dask/bag/tests/test_bag.py::test_to_textfiles[-open] dask/bag/tests/test_bag.py::test_to_textfiles_name_function_preserves_order dask/bag/tests/test_bag.py::test_to_textfiles_name_function_warn dask/bag/tests/test_bag.py::test_to_textfiles_encoding dask/bag/tests/test_bag.py::test_to_textfiles_inputs dask/bag/tests/test_bag.py::test_to_textfiles_endlines dask/bag/tests/test_bag.py::test_string_namespace dask/bag/tests/test_bag.py::test_string_namespace_with_unicode dask/bag/tests/test_bag.py::test_str_empty_split dask/bag/tests/test_bag.py::test_map_with_iterator_function dask/bag/tests/test_bag.py::test_ensure_compute_output_is_concrete dask/bag/tests/test_bag.py::test_bag_class_extend dask/bag/tests/test_bag.py::test_gh715 dask/bag/tests/test_bag.py::test_bag_compute_forward_kwargs dask/bag/tests/test_bag.py::test_to_delayed dask/bag/tests/test_bag.py::test_to_delayed_optimize_graph dask/bag/tests/test_bag.py::test_from_delayed dask/bag/tests/test_bag.py::test_from_delayed_iterator dask/bag/tests/test_bag.py::test_range dask/bag/tests/test_bag.py::test_zip[1] dask/bag/tests/test_bag.py::test_zip[7] dask/bag/tests/test_bag.py::test_zip[10] dask/bag/tests/test_bag.py::test_zip[28] dask/bag/tests/test_bag.py::test_repartition[1-1] dask/bag/tests/test_bag.py::test_repartition[1-2] dask/bag/tests/test_bag.py::test_repartition[1-7] dask/bag/tests/test_bag.py::test_repartition[1-11] dask/bag/tests/test_bag.py::test_repartition[1-23] dask/bag/tests/test_bag.py::test_repartition[2-1] dask/bag/tests/test_bag.py::test_repartition[2-2] dask/bag/tests/test_bag.py::test_repartition[2-7] dask/bag/tests/test_bag.py::test_repartition[2-11] dask/bag/tests/test_bag.py::test_repartition[2-23] dask/bag/tests/test_bag.py::test_repartition[5-1] dask/bag/tests/test_bag.py::test_repartition[5-2] dask/bag/tests/test_bag.py::test_repartition[5-7] dask/bag/tests/test_bag.py::test_repartition[5-11] dask/bag/tests/test_bag.py::test_repartition[5-23] dask/bag/tests/test_bag.py::test_repartition[12-1] dask/bag/tests/test_bag.py::test_repartition[12-2] dask/bag/tests/test_bag.py::test_repartition[12-7] dask/bag/tests/test_bag.py::test_repartition[12-11] dask/bag/tests/test_bag.py::test_repartition[12-23] dask/bag/tests/test_bag.py::test_repartition[23-1] dask/bag/tests/test_bag.py::test_repartition[23-2] dask/bag/tests/test_bag.py::test_repartition[23-7] dask/bag/tests/test_bag.py::test_repartition[23-11] dask/bag/tests/test_bag.py::test_repartition[23-23] dask/bag/tests/test_bag.py::test_repartition_names dask/bag/tests/test_bag.py::test_accumulate dask/bag/tests/test_bag.py::test_groupby_tasks dask/bag/tests/test_bag.py::test_groupby_tasks_names dask/bag/tests/test_bag.py::test_groupby_tasks_2[1000-20-100] dask/bag/tests/test_bag.py::test_groupby_tasks_2[12345-234-1042] dask/bag/tests/test_bag.py::test_groupby_tasks_2[100-1-50] dask/bag/tests/test_bag.py::test_groupby_tasks_3 dask/bag/tests/test_bag.py::test_to_textfiles_empty_partitions dask/bag/tests/test_bag.py::test_reduction_empty dask/bag/tests/test_bag.py::test_reduction_empty_aggregate[1] dask/bag/tests/test_bag.py::test_reduction_empty_aggregate[2] dask/bag/tests/test_bag.py::test_reduction_empty_aggregate[4] dask/bag/tests/test_bag.py::test_reduction_with_non_comparable_objects dask/bag/tests/test_bag.py::test_empty dask/bag/tests/test_bag.py::test_bag_picklable dask/bag/tests/test_bag.py::test_msgpack_unicode dask/bag/tests/test_bag.py::test_bag_with_single_callable dask/bag/tests/test_bag.py::test_optimize_fuse_keys dask/bag/tests/test_bag.py::test_reductions_are_lazy dask/bag/tests/test_bag.py::test_repeated_groupby dask/bag/tests/test_bag.py::test_temporary_directory dask/bag/tests/test_bag.py::test_empty_bag dask/bag/tests/test_bag.py::test_bag_paths dask/bag/tests/test_bag.py::test_map_partitions_arg dask/bag/tests/test_bag.py::test_map_keynames dask/bag/tests/test_bag.py::test_map_releases_element_references_as_soon_as_possible dask/bag/tests/test_bag.py::test_bagged_array_delayed dask/dataframe/io/tests/test_parquet.py::test_parse_pandas_metadata[pandas_metadata0] dask/dataframe/io/tests/test_parquet.py::test_parse_pandas_metadata[pandas_metadata1] dask/dataframe/io/tests/test_parquet.py::test_parse_pandas_metadata[pandas_metadata2] dask/dataframe/io/tests/test_parquet.py::test_parse_pandas_metadata_null_index dask/dataframe/io/tests/test_parquet.py::test_parse_pandas_metadata_duplicate_index_columns dask/dataframe/io/tests/test_parquet.py::test_parse_pandas_metadata_column_with_index_name dask/dataframe/io/tests/test_parquet.py::test_informative_error_messages dask/dataframe/io/tests/test_parquet.py::test_subgraph_getitem dask/dataframe/tests/test_dataframe.py::test_dataframe_doc dask/dataframe/tests/test_dataframe.py::test_dataframe_doc_from_non_pandas dask/dataframe/tests/test_dataframe.py::test_Dataframe dask/dataframe/tests/test_dataframe.py::test_head_tail dask/dataframe/tests/test_dataframe.py::test_head_npartitions dask/dataframe/tests/test_dataframe.py::test_head_npartitions_warn dask/dataframe/tests/test_dataframe.py::test_index_head dask/dataframe/tests/test_dataframe.py::test_Series dask/dataframe/tests/test_dataframe.py::test_Index dask/dataframe/tests/test_dataframe.py::test_Scalar dask/dataframe/tests/test_dataframe.py::test_scalar_raises dask/dataframe/tests/test_dataframe.py::test_attributes dask/dataframe/tests/test_dataframe.py::test_column_names dask/dataframe/tests/test_dataframe.py::test_index_names dask/dataframe/tests/test_dataframe.py::test_rename_columns dask/dataframe/tests/test_dataframe.py::test_rename_series dask/dataframe/tests/test_dataframe.py::test_rename_series_method dask/dataframe/tests/test_dataframe.py::test_rename_series_method_2 dask/dataframe/tests/test_dataframe.py::test_describe_numeric[dask-test_values1] dask/dataframe/tests/test_dataframe.py::test_describe[None-None-None-subset0] dask/dataframe/tests/test_dataframe.py::test_describe[None-None-None-subset2] dask/dataframe/tests/test_dataframe.py::test_describe_empty dask/dataframe/tests/test_dataframe.py::test_describe_for_possibly_unsorted_q dask/dataframe/tests/test_dataframe.py::test_cumulative dask/dataframe/tests/test_dataframe.py::test_cumulative_empty_partitions[func0] dask/dataframe/tests/test_dataframe.py::test_cumulative_empty_partitions[func1] dask/dataframe/tests/test_dataframe.py::test_dropna dask/dataframe/tests/test_dataframe.py::test_clip[2-5] dask/dataframe/tests/test_dataframe.py::test_clip[2.5-3.5] dask/dataframe/tests/test_dataframe.py::test_squeeze dask/dataframe/tests/test_dataframe.py::test_where_mask dask/dataframe/tests/test_dataframe.py::test_map_partitions_multi_argument dask/dataframe/tests/test_dataframe.py::test_map_partitions dask/dataframe/tests/test_dataframe.py::test_map_partitions_type dask/dataframe/tests/test_dataframe.py::test_map_partitions_names dask/dataframe/tests/test_dataframe.py::test_map_partitions_column_info dask/dataframe/tests/test_dataframe.py::test_map_partitions_method_names dask/dataframe/tests/test_dataframe.py::test_map_partitions_propagates_index_metadata dask/dataframe/tests/test_dataframe.py::test_metadata_inference_single_partition_aligned_args dask/dataframe/tests/test_dataframe.py::test_drop_duplicates dask/dataframe/tests/test_dataframe.py::test_drop_duplicates_subset dask/dataframe/tests/test_dataframe.py::test_get_partition dask/dataframe/tests/test_dataframe.py::test_ndim dask/dataframe/tests/test_dataframe.py::test_dtype dask/dataframe/tests/test_dataframe.py::test_value_counts dask/dataframe/tests/test_dataframe.py::test_unique dask/dataframe/tests/test_dataframe.py::test_isin dask/dataframe/tests/test_dataframe.py::test_size dask/dataframe/tests/test_dataframe.py::test_shape dask/dataframe/tests/test_dataframe.py::test_nbytes dask/dataframe/tests/test_dataframe.py::test_quantile[dask-expected1] dask/dataframe/tests/test_dataframe.py::test_quantile_missing[dask] dask/dataframe/tests/test_dataframe.py::test_empty_quantile[dask] dask/dataframe/tests/test_dataframe.py::test_dataframe_quantile[dask-expected1] dask/dataframe/tests/test_dataframe.py::test_quantile_for_possibly_unsorted_q dask/dataframe/tests/test_dataframe.py::test_index dask/dataframe/tests/test_dataframe.py::test_assign dask/dataframe/tests/test_dataframe.py::test_assign_callable dask/dataframe/tests/test_dataframe.py::test_assign_dtypes dask/dataframe/tests/test_dataframe.py::test_map dask/dataframe/tests/test_dataframe.py::test_concat dask/dataframe/tests/test_dataframe.py::test_args dask/dataframe/tests/test_dataframe.py::test_known_divisions dask/dataframe/tests/test_dataframe.py::test_unknown_divisions dask/dataframe/tests/test_dataframe.py::test_with_min_count dask/dataframe/tests/test_dataframe.py::test_align[inner] dask/dataframe/tests/test_dataframe.py::test_align[outer] dask/dataframe/tests/test_dataframe.py::test_align[left] dask/dataframe/tests/test_dataframe.py::test_align[right] dask/dataframe/tests/test_dataframe.py::test_align_axis[inner] dask/dataframe/tests/test_dataframe.py::test_align_axis[outer] dask/dataframe/tests/test_dataframe.py::test_align_axis[left] dask/dataframe/tests/test_dataframe.py::test_align_axis[right] dask/dataframe/tests/test_dataframe.py::test_combine dask/dataframe/tests/test_dataframe.py::test_combine_first dask/dataframe/tests/test_dataframe.py::test_dataframe_picklable dask/dataframe/tests/test_dataframe.py::test_random_partitions dask/dataframe/tests/test_dataframe.py::test_series_round dask/dataframe/tests/test_dataframe.py::test_repartition_divisions dask/dataframe/tests/test_dataframe.py::test_repartition_on_pandas_dataframe dask/dataframe/tests/test_dataframe.py::test_repartition_npartitions[<lambda>0-float-1-1-True] dask/dataframe/tests/test_dataframe.py::test_repartition_npartitions[<lambda>0-float-1-1-False] dask/dataframe/tests/test_dataframe.py::test_repartition_npartitions[<lambda>0-float-1-2-True] dask/dataframe/tests/test_dataframe.py::test_repartition_npartitions[<lambda>0-float-1-2-False] dask/dataframe/tests/test_dataframe.py::test_repartition_npartitions[<lambda>0-float-1-4-True] dask/dataframe/tests/test_dataframe.py::test_repartition_npartitions[<lambda>0-float-1-4-False] dask/dataframe/tests/test_dataframe.py::test_repartition_npartitions[<lambda>0-float-1-5-True] dask/dataframe/tests/test_dataframe.py::test_repartition_npartitions[<lambda>0-float-1-5-False] dask/dataframe/tests/test_dataframe.py::test_repartition_npartitions[<lambda>0-float-2-1-True] dask/dataframe/tests/test_dataframe.py::test_repartition_npartitions[<lambda>0-float-2-1-False] dask/dataframe/tests/test_dataframe.py::test_repartition_npartitions[<lambda>0-float-2-2-True] dask/dataframe/tests/test_dataframe.py::test_repartition_npartitions[<lambda>0-float-2-2-False] dask/dataframe/tests/test_dataframe.py::test_repartition_npartitions[<lambda>0-float-2-4-True] dask/dataframe/tests/test_dataframe.py::test_repartition_npartitions[<lambda>0-float-2-4-False] dask/dataframe/tests/test_dataframe.py::test_repartition_npartitions[<lambda>0-float-2-5-True] dask/dataframe/tests/test_dataframe.py::test_repartition_npartitions[<lambda>0-float-2-5-False] dask/dataframe/tests/test_dataframe.py::test_repartition_npartitions[<lambda>0-float-4-1-True] dask/dataframe/tests/test_dataframe.py::test_repartition_npartitions[<lambda>0-float-4-1-False] dask/dataframe/tests/test_dataframe.py::test_repartition_npartitions[<lambda>0-float-4-2-True] dask/dataframe/tests/test_dataframe.py::test_repartition_npartitions[<lambda>0-float-4-2-False] dask/dataframe/tests/test_dataframe.py::test_repartition_npartitions[<lambda>0-float-4-4-True] dask/dataframe/tests/test_dataframe.py::test_repartition_npartitions[<lambda>0-float-4-4-False] dask/dataframe/tests/test_dataframe.py::test_repartition_npartitions[<lambda>0-float-4-5-True] dask/dataframe/tests/test_dataframe.py::test_repartition_npartitions[<lambda>0-float-4-5-False] dask/dataframe/tests/test_dataframe.py::test_repartition_npartitions[<lambda>0-float-5-1-True] dask/dataframe/tests/test_dataframe.py::test_repartition_npartitions[<lambda>0-float-5-1-False] dask/dataframe/tests/test_dataframe.py::test_repartition_npartitions[<lambda>0-float-5-2-True] dask/dataframe/tests/test_dataframe.py::test_repartition_npartitions[<lambda>0-float-5-2-False] dask/dataframe/tests/test_dataframe.py::test_repartition_npartitions[<lambda>0-float-5-4-True] dask/dataframe/tests/test_dataframe.py::test_repartition_npartitions[<lambda>0-float-5-4-False] dask/dataframe/tests/test_dataframe.py::test_repartition_npartitions[<lambda>0-float-5-5-True] dask/dataframe/tests/test_dataframe.py::test_repartition_npartitions[<lambda>0-float-5-5-False] dask/dataframe/tests/test_dataframe.py::test_repartition_npartitions[<lambda>0-M8[ns]-1-1-True] dask/dataframe/tests/test_dataframe.py::test_repartition_npartitions[<lambda>0-M8[ns]-1-1-False] dask/dataframe/tests/test_dataframe.py::test_repartition_npartitions[<lambda>0-M8[ns]-1-2-True] dask/dataframe/tests/test_dataframe.py::test_repartition_npartitions[<lambda>0-M8[ns]-1-2-False] dask/dataframe/tests/test_dataframe.py::test_repartition_npartitions[<lambda>0-M8[ns]-1-4-True] dask/dataframe/tests/test_dataframe.py::test_repartition_npartitions[<lambda>0-M8[ns]-1-4-False] dask/dataframe/tests/test_dataframe.py::test_repartition_npartitions[<lambda>0-M8[ns]-1-5-True] dask/dataframe/tests/test_dataframe.py::test_repartition_npartitions[<lambda>0-M8[ns]-1-5-False] dask/dataframe/tests/test_dataframe.py::test_repartition_npartitions[<lambda>0-M8[ns]-2-1-True] dask/dataframe/tests/test_dataframe.py::test_repartition_npartitions[<lambda>0-M8[ns]-2-1-False] dask/dataframe/tests/test_dataframe.py::test_repartition_npartitions[<lambda>0-M8[ns]-2-2-True] dask/dataframe/tests/test_dataframe.py::test_repartition_npartitions[<lambda>0-M8[ns]-2-2-False] dask/dataframe/tests/test_dataframe.py::test_repartition_npartitions[<lambda>0-M8[ns]-2-4-True] dask/dataframe/tests/test_dataframe.py::test_repartition_npartitions[<lambda>0-M8[ns]-2-4-False] dask/dataframe/tests/test_dataframe.py::test_repartition_npartitions[<lambda>0-M8[ns]-2-5-True] dask/dataframe/tests/test_dataframe.py::test_repartition_npartitions[<lambda>0-M8[ns]-2-5-False] dask/dataframe/tests/test_dataframe.py::test_repartition_npartitions[<lambda>0-M8[ns]-4-1-True] dask/dataframe/tests/test_dataframe.py::test_repartition_npartitions[<lambda>0-M8[ns]-4-1-False] dask/dataframe/tests/test_dataframe.py::test_repartition_npartitions[<lambda>0-M8[ns]-4-2-True] dask/dataframe/tests/test_dataframe.py::test_repartition_npartitions[<lambda>0-M8[ns]-4-2-False] dask/dataframe/tests/test_dataframe.py::test_repartition_npartitions[<lambda>0-M8[ns]-4-4-True] dask/dataframe/tests/test_dataframe.py::test_repartition_npartitions[<lambda>0-M8[ns]-4-4-False] dask/dataframe/tests/test_dataframe.py::test_repartition_npartitions[<lambda>0-M8[ns]-4-5-True] dask/dataframe/tests/test_dataframe.py::test_repartition_npartitions[<lambda>0-M8[ns]-4-5-False] dask/dataframe/tests/test_dataframe.py::test_repartition_npartitions[<lambda>0-M8[ns]-5-1-True] dask/dataframe/tests/test_dataframe.py::test_repartition_npartitions[<lambda>0-M8[ns]-5-1-False] dask/dataframe/tests/test_dataframe.py::test_repartition_npartitions[<lambda>0-M8[ns]-5-2-True] dask/dataframe/tests/test_dataframe.py::test_repartition_npartitions[<lambda>0-M8[ns]-5-2-False] dask/dataframe/tests/test_dataframe.py::test_repartition_npartitions[<lambda>0-M8[ns]-5-4-True] dask/dataframe/tests/test_dataframe.py::test_repartition_npartitions[<lambda>0-M8[ns]-5-4-False] dask/dataframe/tests/test_dataframe.py::test_repartition_npartitions[<lambda>0-M8[ns]-5-5-True] dask/dataframe/tests/test_dataframe.py::test_repartition_npartitions[<lambda>0-M8[ns]-5-5-False] dask/dataframe/tests/test_dataframe.py::test_repartition_npartitions[<lambda>1-float-1-1-True] dask/dataframe/tests/test_dataframe.py::test_repartition_npartitions[<lambda>1-float-1-1-False] dask/dataframe/tests/test_dataframe.py::test_repartition_npartitions[<lambda>1-float-1-2-True] dask/dataframe/tests/test_dataframe.py::test_repartition_npartitions[<lambda>1-float-1-2-False] dask/dataframe/tests/test_dataframe.py::test_repartition_npartitions[<lambda>1-float-1-4-True] dask/dataframe/tests/test_dataframe.py::test_repartition_npartitions[<lambda>1-float-1-4-False] dask/dataframe/tests/test_dataframe.py::test_repartition_npartitions[<lambda>1-float-1-5-True] dask/dataframe/tests/test_dataframe.py::test_repartition_npartitions[<lambda>1-float-1-5-False] dask/dataframe/tests/test_dataframe.py::test_repartition_npartitions[<lambda>1-float-2-1-True] dask/dataframe/tests/test_dataframe.py::test_repartition_npartitions[<lambda>1-float-2-1-False] dask/dataframe/tests/test_dataframe.py::test_repartition_npartitions[<lambda>1-float-2-2-True] dask/dataframe/tests/test_dataframe.py::test_repartition_npartitions[<lambda>1-float-2-2-False] dask/dataframe/tests/test_dataframe.py::test_repartition_npartitions[<lambda>1-float-2-4-True] dask/dataframe/tests/test_dataframe.py::test_repartition_npartitions[<lambda>1-float-2-4-False] dask/dataframe/tests/test_dataframe.py::test_repartition_npartitions[<lambda>1-float-2-5-True] dask/dataframe/tests/test_dataframe.py::test_repartition_npartitions[<lambda>1-float-2-5-False] dask/dataframe/tests/test_dataframe.py::test_repartition_npartitions[<lambda>1-float-4-1-True] dask/dataframe/tests/test_dataframe.py::test_repartition_npartitions[<lambda>1-float-4-1-False] dask/dataframe/tests/test_dataframe.py::test_repartition_npartitions[<lambda>1-float-4-2-True] dask/dataframe/tests/test_dataframe.py::test_repartition_npartitions[<lambda>1-float-4-2-False] dask/dataframe/tests/test_dataframe.py::test_repartition_npartitions[<lambda>1-float-4-4-True] dask/dataframe/tests/test_dataframe.py::test_repartition_npartitions[<lambda>1-float-4-4-False] dask/dataframe/tests/test_dataframe.py::test_repartition_npartitions[<lambda>1-float-4-5-True] dask/dataframe/tests/test_dataframe.py::test_repartition_npartitions[<lambda>1-float-4-5-False] dask/dataframe/tests/test_dataframe.py::test_repartition_npartitions[<lambda>1-float-5-1-True] dask/dataframe/tests/test_dataframe.py::test_repartition_npartitions[<lambda>1-float-5-1-False] dask/dataframe/tests/test_dataframe.py::test_repartition_npartitions[<lambda>1-float-5-2-True] dask/dataframe/tests/test_dataframe.py::test_repartition_npartitions[<lambda>1-float-5-2-False] dask/dataframe/tests/test_dataframe.py::test_repartition_npartitions[<lambda>1-float-5-4-True] dask/dataframe/tests/test_dataframe.py::test_repartition_npartitions[<lambda>1-float-5-4-False] dask/dataframe/tests/test_dataframe.py::test_repartition_npartitions[<lambda>1-float-5-5-True] dask/dataframe/tests/test_dataframe.py::test_repartition_npartitions[<lambda>1-float-5-5-False] dask/dataframe/tests/test_dataframe.py::test_repartition_npartitions[<lambda>1-M8[ns]-1-1-True] dask/dataframe/tests/test_dataframe.py::test_repartition_npartitions[<lambda>1-M8[ns]-1-1-False] dask/dataframe/tests/test_dataframe.py::test_repartition_npartitions[<lambda>1-M8[ns]-1-2-True] dask/dataframe/tests/test_dataframe.py::test_repartition_npartitions[<lambda>1-M8[ns]-1-2-False] dask/dataframe/tests/test_dataframe.py::test_repartition_npartitions[<lambda>1-M8[ns]-1-4-True] dask/dataframe/tests/test_dataframe.py::test_repartition_npartitions[<lambda>1-M8[ns]-1-4-False] dask/dataframe/tests/test_dataframe.py::test_repartition_npartitions[<lambda>1-M8[ns]-1-5-True] dask/dataframe/tests/test_dataframe.py::test_repartition_npartitions[<lambda>1-M8[ns]-1-5-False] dask/dataframe/tests/test_dataframe.py::test_repartition_npartitions[<lambda>1-M8[ns]-2-1-True] dask/dataframe/tests/test_dataframe.py::test_repartition_npartitions[<lambda>1-M8[ns]-2-1-False] dask/dataframe/tests/test_dataframe.py::test_repartition_npartitions[<lambda>1-M8[ns]-2-2-True] dask/dataframe/tests/test_dataframe.py::test_repartition_npartitions[<lambda>1-M8[ns]-2-2-False] dask/dataframe/tests/test_dataframe.py::test_repartition_npartitions[<lambda>1-M8[ns]-2-4-True] dask/dataframe/tests/test_dataframe.py::test_repartition_npartitions[<lambda>1-M8[ns]-2-4-False] dask/dataframe/tests/test_dataframe.py::test_repartition_npartitions[<lambda>1-M8[ns]-2-5-True] dask/dataframe/tests/test_dataframe.py::test_repartition_npartitions[<lambda>1-M8[ns]-2-5-False] dask/dataframe/tests/test_dataframe.py::test_repartition_npartitions[<lambda>1-M8[ns]-4-1-True] dask/dataframe/tests/test_dataframe.py::test_repartition_npartitions[<lambda>1-M8[ns]-4-1-False] dask/dataframe/tests/test_dataframe.py::test_repartition_npartitions[<lambda>1-M8[ns]-4-2-True] dask/dataframe/tests/test_dataframe.py::test_repartition_npartitions[<lambda>1-M8[ns]-4-2-False] dask/dataframe/tests/test_dataframe.py::test_repartition_npartitions[<lambda>1-M8[ns]-4-4-True] dask/dataframe/tests/test_dataframe.py::test_repartition_npartitions[<lambda>1-M8[ns]-4-4-False] dask/dataframe/tests/test_dataframe.py::test_repartition_npartitions[<lambda>1-M8[ns]-4-5-True] dask/dataframe/tests/test_dataframe.py::test_repartition_npartitions[<lambda>1-M8[ns]-4-5-False] dask/dataframe/tests/test_dataframe.py::test_repartition_npartitions[<lambda>1-M8[ns]-5-1-True] dask/dataframe/tests/test_dataframe.py::test_repartition_npartitions[<lambda>1-M8[ns]-5-1-False] dask/dataframe/tests/test_dataframe.py::test_repartition_npartitions[<lambda>1-M8[ns]-5-2-True] dask/dataframe/tests/test_dataframe.py::test_repartition_npartitions[<lambda>1-M8[ns]-5-2-False] dask/dataframe/tests/test_dataframe.py::test_repartition_npartitions[<lambda>1-M8[ns]-5-4-True] dask/dataframe/tests/test_dataframe.py::test_repartition_npartitions[<lambda>1-M8[ns]-5-4-False] dask/dataframe/tests/test_dataframe.py::test_repartition_npartitions[<lambda>1-M8[ns]-5-5-True] dask/dataframe/tests/test_dataframe.py::test_repartition_npartitions[<lambda>1-M8[ns]-5-5-False] dask/dataframe/tests/test_dataframe.py::test_repartition_partition_size[<lambda>0-1kiB-2-True] dask/dataframe/tests/test_dataframe.py::test_repartition_partition_size[<lambda>0-1kiB-2-False] dask/dataframe/tests/test_dataframe.py::test_repartition_partition_size[<lambda>0-1kiB-5-True] dask/dataframe/tests/test_dataframe.py::test_repartition_partition_size[<lambda>0-1kiB-5-False] dask/dataframe/tests/test_dataframe.py::test_repartition_partition_size[<lambda>0-379-2-True] dask/dataframe/tests/test_dataframe.py::test_repartition_partition_size[<lambda>0-379-2-False] dask/dataframe/tests/test_dataframe.py::test_repartition_partition_size[<lambda>0-379-5-True] dask/dataframe/tests/test_dataframe.py::test_repartition_partition_size[<lambda>0-379-5-False] dask/dataframe/tests/test_dataframe.py::test_repartition_partition_size[<lambda>1-1kiB-2-True] dask/dataframe/tests/test_dataframe.py::test_repartition_partition_size[<lambda>1-1kiB-2-False] dask/dataframe/tests/test_dataframe.py::test_repartition_partition_size[<lambda>1-1kiB-5-True] dask/dataframe/tests/test_dataframe.py::test_repartition_partition_size[<lambda>1-1kiB-5-False] dask/dataframe/tests/test_dataframe.py::test_repartition_partition_size[<lambda>1-379-2-True] dask/dataframe/tests/test_dataframe.py::test_repartition_partition_size[<lambda>1-379-2-False] dask/dataframe/tests/test_dataframe.py::test_repartition_partition_size[<lambda>1-379-5-True] dask/dataframe/tests/test_dataframe.py::test_repartition_partition_size[<lambda>1-379-5-False] dask/dataframe/tests/test_dataframe.py::test_iter_chunks dask/dataframe/tests/test_dataframe.py::test_repartition_npartitions_same_limits dask/dataframe/tests/test_dataframe.py::test_repartition_npartitions_numeric_edge_case dask/dataframe/tests/test_dataframe.py::test_repartition_object_index dask/dataframe/tests/test_dataframe.py::test_repartition_freq_divisions dask/dataframe/tests/test_dataframe.py::test_repartition_freq_errors dask/dataframe/tests/test_dataframe.py::test_repartition_freq_month dask/dataframe/tests/test_dataframe.py::test_repartition_input_errors dask/dataframe/tests/test_dataframe.py::test_fillna dask/dataframe/tests/test_dataframe.py::test_fillna_duplicate_index dask/dataframe/tests/test_dataframe.py::test_fillna_multi_dataframe dask/dataframe/tests/test_dataframe.py::test_ffill_bfill dask/dataframe/tests/test_dataframe.py::test_fillna_series_types dask/dataframe/tests/test_dataframe.py::test_sample dask/dataframe/tests/test_dataframe.py::test_sample_without_replacement dask/dataframe/tests/test_dataframe.py::test_sample_raises dask/dataframe/tests/test_dataframe.py::test_empty_max dask/dataframe/tests/test_dataframe.py::test_select_dtypes[include0-None] dask/dataframe/tests/test_dataframe.py::test_select_dtypes[None-exclude1] dask/dataframe/tests/test_dataframe.py::test_select_dtypes[include2-exclude2] dask/dataframe/tests/test_dataframe.py::test_select_dtypes[include3-None] dask/dataframe/tests/test_dataframe.py::test_deterministic_apply_concat_apply_names dask/dataframe/tests/test_dataframe.py::test_aca_meta_infer dask/dataframe/tests/test_dataframe.py::test_aca_split_every dask/dataframe/tests/test_dataframe.py::test_reduction_method dask/dataframe/tests/test_dataframe.py::test_reduction_method_split_every dask/dataframe/tests/test_dataframe.py::test_pipe dask/dataframe/tests/test_dataframe.py::test_gh_517 dask/dataframe/tests/test_dataframe.py::test_drop_axis_1 dask/dataframe/tests/test_dataframe.py::test_gh580 dask/dataframe/tests/test_dataframe.py::test_rename_dict dask/dataframe/tests/test_dataframe.py::test_rename_function dask/dataframe/tests/test_dataframe.py::test_rename_index dask/dataframe/tests/test_dataframe.py::test_to_frame dask/dataframe/tests/test_dataframe.py::test_to_dask_array_raises[False0] dask/dataframe/tests/test_dataframe.py::test_to_dask_array_raises[False1] dask/dataframe/tests/test_dataframe.py::test_to_dask_array_unknown[False0] dask/dataframe/tests/test_dataframe.py::test_to_dask_array_unknown[False1] dask/dataframe/tests/test_dataframe.py::test_to_dask_array[False0-lengths0] dask/dataframe/tests/test_dataframe.py::test_to_dask_array[False0-True] dask/dataframe/tests/test_dataframe.py::test_to_dask_array[False1-lengths0] dask/dataframe/tests/test_dataframe.py::test_to_dask_array[False1-True] dask/dataframe/tests/test_dataframe.py::test_apply dask/dataframe/tests/test_dataframe.py::test_applymap dask/dataframe/tests/test_dataframe.py::test_abs dask/dataframe/tests/test_dataframe.py::test_round dask/dataframe/tests/test_dataframe.py::test_cov dask/dataframe/tests/test_dataframe.py::test_corr dask/dataframe/tests/test_dataframe.py::test_corr_same_name dask/dataframe/tests/test_dataframe.py::test_cov_corr_meta dask/dataframe/tests/test_dataframe.py::test_cov_corr_mixed dask/dataframe/tests/test_dataframe.py::test_autocorr dask/dataframe/tests/test_dataframe.py::test_apply_infer_columns dask/dataframe/tests/test_dataframe.py::test_index_time_properties dask/dataframe/tests/test_dataframe.py::test_nlargest_nsmallest dask/dataframe/tests/test_dataframe.py::test_reset_index dask/dataframe/tests/test_dataframe.py::test_dataframe_compute_forward_kwargs dask/dataframe/tests/test_dataframe.py::test_series_iteritems dask/dataframe/tests/test_dataframe.py::test_series_iter dask/dataframe/tests/test_dataframe.py::test_dataframe_iterrows dask/dataframe/tests/test_dataframe.py::test_dataframe_itertuples dask/dataframe/tests/test_dataframe.py::test_dataframe_itertuples_with_index_false dask/dataframe/tests/test_dataframe.py::test_dataframe_itertuples_with_name_none dask/dataframe/tests/test_dataframe.py::test_astype dask/dataframe/tests/test_dataframe.py::test_groupby_callable dask/dataframe/tests/test_dataframe.py::test_methods_tokenize_differently dask/dataframe/tests/test_dataframe.py::test_info dask/dataframe/tests/test_dataframe.py::test_groupby_multilevel_info dask/dataframe/tests/test_dataframe.py::test_categorize_info dask/dataframe/tests/test_dataframe.py::test_gh_1301 dask/dataframe/tests/test_dataframe.py::test_timeseries_sorted dask/dataframe/tests/test_dataframe.py::test_column_assignment dask/dataframe/tests/test_dataframe.py::test_array_assignment dask/dataframe/tests/test_dataframe.py::test_columns_assignment dask/dataframe/tests/test_dataframe.py::test_attribute_assignment dask/dataframe/tests/test_dataframe.py::test_setitem_triggering_realign dask/dataframe/tests/test_dataframe.py::test_inplace_operators dask/dataframe/tests/test_dataframe.py::test_idxmaxmin[idx0-True] dask/dataframe/tests/test_dataframe.py::test_idxmaxmin[idx0-False] dask/dataframe/tests/test_dataframe.py::test_idxmaxmin[idx1-True] dask/dataframe/tests/test_dataframe.py::test_idxmaxmin[idx1-False] dask/dataframe/tests/test_dataframe.py::test_idxmaxmin_empty_partitions dask/dataframe/tests/test_dataframe.py::test_getitem_meta dask/dataframe/tests/test_dataframe.py::test_getitem_multilevel dask/dataframe/tests/test_dataframe.py::test_getitem_string_subclass dask/dataframe/tests/test_dataframe.py::test_getitem_column_types[list] dask/dataframe/tests/test_dataframe.py::test_getitem_column_types[array] dask/dataframe/tests/test_dataframe.py::test_getitem_column_types[Series] dask/dataframe/tests/test_dataframe.py::test_getitem_column_types[Index] dask/dataframe/tests/test_dataframe.py::test_ipython_completion dask/dataframe/tests/test_dataframe.py::test_diff dask/dataframe/tests/test_dataframe.py::test_shift dask/dataframe/tests/test_dataframe.py::test_shift_with_freq_DatetimeIndex[B-False] dask/dataframe/tests/test_dataframe.py::test_shift_with_freq_DatetimeIndex[D-True] dask/dataframe/tests/test_dataframe.py::test_shift_with_freq_DatetimeIndex[H-True] dask/dataframe/tests/test_dataframe.py::test_shift_with_freq_PeriodIndex[B-False] dask/dataframe/tests/test_dataframe.py::test_shift_with_freq_PeriodIndex[D-True] dask/dataframe/tests/test_dataframe.py::test_shift_with_freq_PeriodIndex[H-True] dask/dataframe/tests/test_dataframe.py::test_shift_with_freq_TimedeltaIndex dask/dataframe/tests/test_dataframe.py::test_shift_with_freq_errors dask/dataframe/tests/test_dataframe.py::test_first_and_last[last] dask/dataframe/tests/test_dataframe.py::test_hash_split_unique[None-2-1] dask/dataframe/tests/test_dataframe.py::test_hash_split_unique[None-2-4] dask/dataframe/tests/test_dataframe.py::test_hash_split_unique[None-2-20] dask/dataframe/tests/test_dataframe.py::test_hash_split_unique[None-5-1] dask/dataframe/tests/test_dataframe.py::test_hash_split_unique[None-5-4] dask/dataframe/tests/test_dataframe.py::test_hash_split_unique[None-5-20] dask/dataframe/tests/test_dataframe.py::test_hash_split_unique[1-2-1] dask/dataframe/tests/test_dataframe.py::test_hash_split_unique[1-2-4] dask/dataframe/tests/test_dataframe.py::test_hash_split_unique[1-2-20] dask/dataframe/tests/test_dataframe.py::test_hash_split_unique[1-5-1] dask/dataframe/tests/test_dataframe.py::test_hash_split_unique[1-5-4] dask/dataframe/tests/test_dataframe.py::test_hash_split_unique[1-5-20] dask/dataframe/tests/test_dataframe.py::test_hash_split_unique[5-2-1] dask/dataframe/tests/test_dataframe.py::test_hash_split_unique[5-2-4] dask/dataframe/tests/test_dataframe.py::test_hash_split_unique[5-2-20] dask/dataframe/tests/test_dataframe.py::test_hash_split_unique[5-5-1] dask/dataframe/tests/test_dataframe.py::test_hash_split_unique[5-5-4] dask/dataframe/tests/test_dataframe.py::test_hash_split_unique[5-5-20] dask/dataframe/tests/test_dataframe.py::test_hash_split_unique[20-2-1] dask/dataframe/tests/test_dataframe.py::test_hash_split_unique[20-2-4] dask/dataframe/tests/test_dataframe.py::test_hash_split_unique[20-2-20] dask/dataframe/tests/test_dataframe.py::test_hash_split_unique[20-5-1] dask/dataframe/tests/test_dataframe.py::test_hash_split_unique[20-5-4] dask/dataframe/tests/test_dataframe.py::test_hash_split_unique[20-5-20] dask/dataframe/tests/test_dataframe.py::test_split_out_drop_duplicates[None] dask/dataframe/tests/test_dataframe.py::test_split_out_drop_duplicates[2] dask/dataframe/tests/test_dataframe.py::test_split_out_value_counts[None] dask/dataframe/tests/test_dataframe.py::test_split_out_value_counts[2] dask/dataframe/tests/test_dataframe.py::test_values dask/dataframe/tests/test_dataframe.py::test_copy dask/dataframe/tests/test_dataframe.py::test_del dask/dataframe/tests/test_dataframe.py::test_memory_usage[True-True] dask/dataframe/tests/test_dataframe.py::test_memory_usage[True-False] dask/dataframe/tests/test_dataframe.py::test_memory_usage[False-True] dask/dataframe/tests/test_dataframe.py::test_memory_usage[False-False] dask/dataframe/tests/test_dataframe.py::test_memory_usage_per_partition[True-True] dask/dataframe/tests/test_dataframe.py::test_memory_usage_per_partition[True-False] dask/dataframe/tests/test_dataframe.py::test_memory_usage_per_partition[False-True] dask/dataframe/tests/test_dataframe.py::test_memory_usage_per_partition[False-False] dask/dataframe/tests/test_dataframe.py::test_dataframe_reductions_arithmetic[sum] dask/dataframe/tests/test_dataframe.py::test_dataframe_reductions_arithmetic[mean] dask/dataframe/tests/test_dataframe.py::test_dataframe_reductions_arithmetic[std] dask/dataframe/tests/test_dataframe.py::test_dataframe_reductions_arithmetic[var] dask/dataframe/tests/test_dataframe.py::test_dataframe_reductions_arithmetic[count] dask/dataframe/tests/test_dataframe.py::test_dataframe_reductions_arithmetic[min] dask/dataframe/tests/test_dataframe.py::test_dataframe_reductions_arithmetic[max] dask/dataframe/tests/test_dataframe.py::test_dataframe_reductions_arithmetic[idxmin] dask/dataframe/tests/test_dataframe.py::test_dataframe_reductions_arithmetic[idxmax] dask/dataframe/tests/test_dataframe.py::test_dataframe_reductions_arithmetic[prod] dask/dataframe/tests/test_dataframe.py::test_dataframe_reductions_arithmetic[all] dask/dataframe/tests/test_dataframe.py::test_dataframe_reductions_arithmetic[sem] dask/dataframe/tests/test_dataframe.py::test_dataframe_mode dask/dataframe/tests/test_dataframe.py::test_to_datetime dask/dataframe/tests/test_dataframe.py::test_to_timedelta dask/dataframe/tests/test_dataframe.py::test_isna[values0] dask/dataframe/tests/test_dataframe.py::test_isna[values1] dask/dataframe/tests/test_dataframe.py::test_slice_on_filtered_boundary[0] dask/dataframe/tests/test_dataframe.py::test_slice_on_filtered_boundary[9] dask/dataframe/tests/test_dataframe.py::test_boundary_slice_nonmonotonic dask/dataframe/tests/test_dataframe.py::test_boundary_slice_empty dask/dataframe/tests/test_dataframe.py::test_with_boundary[-1-None-False-False-drop0] dask/dataframe/tests/test_dataframe.py::test_with_boundary[-1-None-False-True-drop1] dask/dataframe/tests/test_dataframe.py::test_with_boundary[None-3-False-False-drop2] dask/dataframe/tests/test_dataframe.py::test_with_boundary[None-3-True-False-drop3] dask/dataframe/tests/test_dataframe.py::test_with_boundary[-0.5-None-False-False-drop4] dask/dataframe/tests/test_dataframe.py::test_with_boundary[-0.5-None-False-True-drop5] dask/dataframe/tests/test_dataframe.py::test_with_boundary[-1.5-None-False-True-drop6] dask/dataframe/tests/test_dataframe.py::test_with_boundary[None-3.5-False-False-drop7] dask/dataframe/tests/test_dataframe.py::test_with_boundary[None-3.5-True-False-drop8] dask/dataframe/tests/test_dataframe.py::test_with_boundary[None-2.5-False-False-drop9] dask/dataframe/tests/test_dataframe.py::test_boundary_slice_same[index0-0-9] dask/dataframe/tests/test_dataframe.py::test_boundary_slice_same[index1--1-None] dask/dataframe/tests/test_dataframe.py::test_boundary_slice_same[index2-None-10] dask/dataframe/tests/test_dataframe.py::test_boundary_slice_same[index3-None-None] dask/dataframe/tests/test_dataframe.py::test_boundary_slice_same[index4--1-None] dask/dataframe/tests/test_dataframe.py::test_boundary_slice_same[index5-None-2] dask/dataframe/tests/test_dataframe.py::test_boundary_slice_same[index6--2-3] dask/dataframe/tests/test_dataframe.py::test_boundary_slice_same[index7-None-None] dask/dataframe/tests/test_dataframe.py::test_boundary_slice_same[index8-left8-None] dask/dataframe/tests/test_dataframe.py::test_boundary_slice_same[index9-None-right9] dask/dataframe/tests/test_dataframe.py::test_boundary_slice_same[index10-left10-None] dask/dataframe/tests/test_dataframe.py::test_boundary_slice_same[index11-None-right11] dask/dataframe/tests/test_dataframe.py::test_better_errors_object_reductions dask/dataframe/tests/test_dataframe.py::test_sample_empty_partitions dask/dataframe/tests/test_dataframe.py::test_coerce dask/dataframe/tests/test_dataframe.py::test_bool dask/dataframe/tests/test_dataframe.py::test_cumulative_multiple_columns dask/dataframe/tests/test_dataframe.py::test_map_partition_array[asarray] dask/dataframe/tests/test_dataframe.py::test_map_partition_array[func1] dask/dataframe/tests/test_dataframe.py::test_mixed_dask_array_operations dask/dataframe/tests/test_dataframe.py::test_mixed_dask_array_operations_errors dask/dataframe/tests/test_dataframe.py::test_mixed_dask_array_multi_dimensional dask/dataframe/tests/test_dataframe.py::test_meta_raises dask/dataframe/tests/test_dataframe.py::test_map_partitions_delays_large_inputs dask/dataframe/tests/test_dataframe.py::test_partitions_indexer dask/dataframe/tests/test_dataframe.py::test_mod_eq dask/dataframe/tests/test_dataframe.py::test_setitem dask/dataframe/tests/test_dataframe.py::test_broadcast dask/dataframe/tests/test_dataframe.py::test_scalar_with_array dask/dataframe/tests/test_dataframe.py::test_has_parallel_type dask/dataframe/tests/test_dataframe.py::test_meta_error_message dask/dataframe/tests/test_dataframe.py::test_assign_index dask/dataframe/tests/test_dataframe.py::test_index_divisions dask/dataframe/tests/test_dataframe.py::test_replace dask/dataframe/tests/test_dataframe.py::test_map_partitions_delays_lists dask/dataframe/tests/test_dataframe.py::test_dtype_cast dask/dataframe/tests/test_dataframe.py::test_series_map[False-False-1-1] dask/dataframe/tests/test_dataframe.py::test_series_map[False-False-1-4] dask/dataframe/tests/test_dataframe.py::test_series_map[False-False-3-1] dask/dataframe/tests/test_dataframe.py::test_series_map[False-False-3-4] dask/dataframe/tests/test_dataframe.py::test_series_map[False-True-1-1] dask/dataframe/tests/test_dataframe.py::test_series_map[False-True-1-4] dask/dataframe/tests/test_dataframe.py::test_series_map[False-True-3-1] dask/dataframe/tests/test_dataframe.py::test_series_map[False-True-3-4] dask/dataframe/tests/test_dataframe.py::test_series_map[True-False-1-1] dask/dataframe/tests/test_dataframe.py::test_series_map[True-False-1-4] dask/dataframe/tests/test_dataframe.py::test_series_map[True-False-3-1] dask/dataframe/tests/test_dataframe.py::test_series_map[True-False-3-4] dask/dataframe/tests/test_dataframe.py::test_series_map[True-True-1-1] dask/dataframe/tests/test_dataframe.py::test_series_map[True-True-1-4] dask/dataframe/tests/test_dataframe.py::test_series_map[True-True-3-1] dask/dataframe/tests/test_dataframe.py::test_series_map[True-True-3-4] dask/dataframe/tests/test_dataframe.py::test_dataframe_explode dask/dataframe/tests/test_dataframe.py::test_series_explode dask/dataframe/tests/test_dataframe.py::test_pop dask/dataframe/tests/test_dataframe.py::test_simple_map_partitions dask/dataframe/tests/test_dataframe.py::test_iter dask/dataframe/tests/test_dataframe.py::test_dataframe_groupby_agg_empty_partitions dask/dataframe/tests/test_optimize_dataframe.py::test_fuse_ave_width dask/dataframe/tests/test_optimize_dataframe.py::test_optimize_blockwise dask/tests/test_config.py::test_canonical_name dask/tests/test_config.py::test_update dask/tests/test_config.py::test_merge dask/tests/test_config.py::test_collect_yaml_paths dask/tests/test_config.py::test_collect_yaml_dir dask/tests/test_config.py::test_env dask/tests/test_config.py::test_collect dask/tests/test_config.py::test_collect_env_none dask/tests/test_config.py::test_get dask/tests/test_config.py::test_ensure_file dask/tests/test_config.py::test_set dask/tests/test_config.py::test_set_kwargs dask/tests/test_config.py::test_set_nested dask/tests/test_config.py::test_set_hard_to_copyables dask/tests/test_config.py::test_ensure_file_directory[True] dask/tests/test_config.py::test_ensure_file_directory[False] dask/tests/test_config.py::test_ensure_file_defaults_to_DASK_CONFIG_directory dask/tests/test_config.py::test_rename dask/tests/test_config.py::test_refresh dask/tests/test_config.py::test_expand_environment_variables[1-10] dask/tests/test_config.py::test_expand_environment_variables[1-11] dask/tests/test_config.py::test_expand_environment_variables[$FOO-foo] dask/tests/test_config.py::test_expand_environment_variables[inp3-out3] dask/tests/test_config.py::test_expand_environment_variables[inp4-out4] dask/tests/test_config.py::test_expand_environment_variables[inp5-out5] dask/tests/test_config.py::test_expand_environment_variables[inp6-out6] dask/tests/test_config.py::test_expand_environment_variables[inp7-out7] dask/tests/test_config.py::test_env_var_canonical_name dask/tests/test_config.py::test_get_set_canonical_name dask/tests/test_config.py::test_get_set_roundtrip[custom_key] dask/tests/test_config.py::test_get_set_roundtrip[custom-key] dask/tests/test_config.py::test_merge_None_to_dict dask/tests/test_config.py::test_core_file dask/tests/test_optimization.py::test_cull dask/tests/test_optimization.py::test_fuse dask/tests/test_optimization.py::test_fuse_keys dask/tests/test_optimization.py::test_inline dask/tests/test_optimization.py::test_inline_functions dask/tests/test_optimization.py::test_inline_ignores_curries_and_partials dask/tests/test_optimization.py::test_inline_functions_non_hashable dask/tests/test_optimization.py::test_inline_doesnt_shrink_fast_functions_at_top dask/tests/test_optimization.py::test_inline_traverses_lists dask/tests/test_optimization.py::test_inline_functions_protects_output_keys dask/tests/test_optimization.py::test_functions_of dask/tests/test_optimization.py::test_inline_cull_dependencies dask/tests/test_optimization.py::test_fuse_reductions_single_input dask/tests/test_optimization.py::test_fuse_stressed dask/tests/test_optimization.py::test_fuse_reductions_multiple_input dask/tests/test_optimization.py::test_SubgraphCallable dask/tests/test_optimization.py::test_fuse_subgraphs dask/tests/test_optimization.py::test_fuse_subgraphs_linear_chains_of_duplicate_deps dask/tests/test_optimization.py::test_dont_fuse_numpy_arrays
dask/optimization.py docs/source/delayed.rst dask/dataframe/optimize.py dask/array/random.py dask/dataframe/core.py dask/dask.yaml dask/dataframe/shuffle.py dask/dataframe/io/parquet/arrow.py dask/dataframe/io/parquet/fastparquet.py docs/source/setup/custom-startup.rst dask/config.py dask/dataframe/utils.py dask/array/creation.py dask/utils.py
dask/optimization.py docs/source/delayed.rst dask/dataframe/optimize.py dask/array/random.py dask/dataframe/core.py dask/dask.yaml dask/dataframe/shuffle.py dask/dataframe/io/parquet/arrow.py dask/dataframe/io/parquet/fastparquet.py docs/source/setup/custom-startup.rst dask/config.py dask/dataframe/utils.py dask/array/creation.py dask/utils.py