Repository: dask/dask

Issue:
un-numpy-like behavior of `da.pad` with stats 
`da.pad` with `mode` set to a statistic (e.g., `minimum`) a) doesn't match numpy and b) gives a confusing error message when it fails. The source of this problem is how `da.pad` tries to calculate the `stat_length` parameter when that kwarg is not specified.  

If I have the time, I can put together a PR to fix this. It should not be too hard. The problem is this line: https://github.com/dask/dask/blob/master/dask/array/creation.py#L1007, which blindly uses `args[0]` as `stat_length` without an earlier check for whether `stat_length` has been supplied as a `kwarg`. 

MRE: 
```
import dask.array as da
# this works:
arr_2d = da.zeros((1,1))
da.pad(arr_2d, pad_width=1, mode='minimum')

# error: `TypeError: `pad_value` must be composed of integral typed values.`
arr_3d  = da.zeros((1,1,1))
da.pad(arr_3d, pad_width=1, mode='minimum')

# In numpy, pad with mode=$statistic uses an extra kwarg called `stat_length, which defaults to None. 
# In `da.pad`, passing stat_length=None doesn't work. This produces the same confusing error about the non-integral type of the pad_value.
da.pad(arr_3d, pad_width=1, mode='minimum', stat_length=None)

# this gives equivalent behavior to np.pad with stat_length=None:
da.pad(arr_3d, pad_width=1, mode='minimum', stat_length=arr_3d.shape)
```


Discussion:
mark-boer: It seems that `mode = 'linear_ramp'` also has issues in Python 3.6.
quasiben: Thanks @mark-boer!  I see this is your first time contributing, welcome!  For 3/4, are the test failing ?
mark-boer: Yes if you uncomment out these lines: 

```
        # "mean", TODO bug: dask changes the dtype to float
        "minimum",
        # "reflect", TODO bug: reflect, symmetric and wrap do not work when the pad_width is larger than the dimension
        # "symmetric",
        # "wrap", 
```
martindurant: (fsspec issues should have gone away, sorry about that. Will do bugfix release.)
TomAugspurger: @mark-boer for now I'd recommend splitting out the things that do work from the things that don't (so remove the commented test cases) and keep this PR focused on one issue. That'll make it easier to review and merging won't be held up by unrelated issues. 
mark-boer: > so remove the commented test cases
Shall i just mark them xfail, with a link to a github issue?

I will check why python 3.6 failed later this week, maybe numpy changed something between versions 1.15 and 1.17.
mark-boer: I think this is about ready to merge, if you agree with the way I skip the tests with a link to the github issue. I might create a pull request that fixes point 3 or 4 later.

Patch:
diff --git a/dask/array/creation.py b/dask/array/creation.py
index 84a589879..2ca9ddbfd 100644
--- a/dask/array/creation.py
+++ b/dask/array/creation.py
@@ -763,7 +763,9 @@ def repeat(a, repeats, axis=None):
     elif not 0 <= axis <= a.ndim - 1:
         raise ValueError("axis(=%d) out of bounds" % axis)
 
-    if repeats == 1:
+    if repeats == 0:
+        return a[tuple(slice(None) if d != axis else slice(0) for d in range(a.ndim))]
+    elif repeats == 1:
         return a
 
     cchunks = cached_cumsum(a.chunks[axis], initial_zero=True)
@@ -835,7 +837,7 @@ def expand_pad_value(array, pad_value):
         and len(pad_value) == 2
         and all(isinstance(pw, Number) for pw in pad_value)
     ):
-        pad_value = tuple((pad_value[0], pad_value[1]) for _ in range(array.ndim))
+        pad_value = array.ndim * (tuple(pad_value),)
     elif (
         isinstance(pad_value, Sequence)
         and len(pad_value) == array.ndim
@@ -843,7 +845,15 @@ def expand_pad_value(array, pad_value):
         and all((len(pw) == 2) for pw in pad_value)
         and all(all(isinstance(w, Number) for w in pw) for pw in pad_value)
     ):
-        pad_value = tuple((pw[0], pw[1]) for pw in pad_value)
+        pad_value = tuple(tuple(pw) for pw in pad_value)
+    elif (
+        isinstance(pad_value, Sequence)
+        and len(pad_value) == 1
+        and isinstance(pad_value[0], Sequence)
+        and len(pad_value[0]) == 2
+        and all(isinstance(pw, Number) for pw in pad_value[0])
+    ):
+        pad_value = array.ndim * (tuple(pad_value[0]),)
     else:
         raise TypeError("`pad_value` must be composed of integral typed values.")
 
@@ -1117,7 +1127,7 @@ def pad(array, pad_width, mode, **kwargs):
     pad_width = expand_pad_value(array, pad_width)
 
     if mode in ["maximum", "mean", "median", "minimum"]:
-        kwargs.setdefault("stat_length", array.shape)
+        kwargs.setdefault("stat_length", tuple((n, n) for n in array.shape))
     elif mode == "constant":
         kwargs.setdefault("constant_values", 0)
     elif mode == "linear_ramp":
diff --git a/dask/array/random.py b/dask/array/random.py
index ed19cfc4b..e4e445941 100644
--- a/dask/array/random.py
+++ b/dask/array/random.py
@@ -196,21 +196,21 @@ class RandomState(object):
         graph = HighLevelGraph.from_collections(name, dsk, dependencies=dependencies)
         return Array(graph, name, chunks + extra_chunks, meta=meta)
 
-    @derived_from(np.random.RandomState)
+    @derived_from(np.random.RandomState, skipblocks=1)
     def beta(self, a, b, size=None, chunks="auto", **kwargs):
         return self._wrap("beta", a, b, size=size, chunks=chunks, **kwargs)
 
-    @derived_from(np.random.RandomState)
+    @derived_from(np.random.RandomState, skipblocks=1)
     def binomial(self, n, p, size=None, chunks="auto", **kwargs):
         return self._wrap("binomial", n, p, size=size, chunks=chunks, **kwargs)
 
-    @derived_from(np.random.RandomState)
+    @derived_from(np.random.RandomState, skipblocks=1)
     def chisquare(self, df, size=None, chunks="auto", **kwargs):
         return self._wrap("chisquare", df, size=size, chunks=chunks, **kwargs)
 
     with ignoring(AttributeError):
 
-        @derived_from(np.random.RandomState)
+        @derived_from(np.random.RandomState, skipblocks=1)
         def choice(self, a, size=None, replace=True, p=None, chunks="auto"):
             dependencies = []
             # Normalize and validate `a`
@@ -282,52 +282,52 @@ class RandomState(object):
             )
             return Array(graph, name, chunks, dtype=dtype)
 
-    # @derived_from(np.random.RandomState)
+    # @derived_from(np.random.RandomState, skipblocks=1)
     # def dirichlet(self, alpha, size=None, chunks="auto"):
 
-    @derived_from(np.random.RandomState)
+    @derived_from(np.random.RandomState, skipblocks=1)
     def exponential(self, scale=1.0, size=None, chunks="auto", **kwargs):
         return self._wrap("exponential", scale, size=size, chunks=chunks, **kwargs)
 
-    @derived_from(np.random.RandomState)
+    @derived_from(np.random.RandomState, skipblocks=1)
     def f(self, dfnum, dfden, size=None, chunks="auto", **kwargs):
         return self._wrap("f", dfnum, dfden, size=size, chunks=chunks, **kwargs)
 
-    @derived_from(np.random.RandomState)
+    @derived_from(np.random.RandomState, skipblocks=1)
     def gamma(self, shape, scale=1.0, size=None, chunks="auto", **kwargs):
         return self._wrap("gamma", shape, scale, size=size, chunks=chunks, **kwargs)
 
-    @derived_from(np.random.RandomState)
+    @derived_from(np.random.RandomState, skipblocks=1)
     def geometric(self, p, size=None, chunks="auto", **kwargs):
         return self._wrap("geometric", p, size=size, chunks=chunks, **kwargs)
 
-    @derived_from(np.random.RandomState)
+    @derived_from(np.random.RandomState, skipblocks=1)
     def gumbel(self, loc=0.0, scale=1.0, size=None, chunks="auto", **kwargs):
         return self._wrap("gumbel", loc, scale, size=size, chunks=chunks, **kwargs)
 
-    @derived_from(np.random.RandomState)
+    @derived_from(np.random.RandomState, skipblocks=1)
     def hypergeometric(self, ngood, nbad, nsample, size=None, chunks="auto", **kwargs):
         return self._wrap(
             "hypergeometric", ngood, nbad, nsample, size=size, chunks=chunks, **kwargs
         )
 
-    @derived_from(np.random.RandomState)
+    @derived_from(np.random.RandomState, skipblocks=1)
     def laplace(self, loc=0.0, scale=1.0, size=None, chunks="auto", **kwargs):
         return self._wrap("laplace", loc, scale, size=size, chunks=chunks, **kwargs)
 
-    @derived_from(np.random.RandomState)
+    @derived_from(np.random.RandomState, skipblocks=1)
     def logistic(self, loc=0.0, scale=1.0, size=None, chunks="auto", **kwargs):
         return self._wrap("logistic", loc, scale, size=size, chunks=chunks, **kwargs)
 
-    @derived_from(np.random.RandomState)
+    @derived_from(np.random.RandomState, skipblocks=1)
     def lognormal(self, mean=0.0, sigma=1.0, size=None, chunks="auto", **kwargs):
         return self._wrap("lognormal", mean, sigma, size=size, chunks=chunks, **kwargs)
 
-    @derived_from(np.random.RandomState)
+    @derived_from(np.random.RandomState, skipblocks=1)
     def logseries(self, p, size=None, chunks="auto", **kwargs):
         return self._wrap("logseries", p, size=size, chunks=chunks, **kwargs)
 
-    @derived_from(np.random.RandomState)
+    @derived_from(np.random.RandomState, skipblocks=1)
     def multinomial(self, n, pvals, size=None, chunks="auto", **kwargs):
         return self._wrap(
             "multinomial",
@@ -338,31 +338,31 @@ class RandomState(object):
             extra_chunks=((len(pvals),),),
         )
 
-    @derived_from(np.random.RandomState)
+    @derived_from(np.random.RandomState, skipblocks=1)
     def negative_binomial(self, n, p, size=None, chunks="auto", **kwargs):
         return self._wrap("negative_binomial", n, p, size=size, chunks=chunks, **kwargs)
 
-    @derived_from(np.random.RandomState)
+    @derived_from(np.random.RandomState, skipblocks=1)
     def noncentral_chisquare(self, df, nonc, size=None, chunks="auto", **kwargs):
         return self._wrap(
             "noncentral_chisquare", df, nonc, size=size, chunks=chunks, **kwargs
         )
 
-    @derived_from(np.random.RandomState)
+    @derived_from(np.random.RandomState, skipblocks=1)
     def noncentral_f(self, dfnum, dfden, nonc, size=None, chunks="auto", **kwargs):
         return self._wrap(
             "noncentral_f", dfnum, dfden, nonc, size=size, chunks=chunks, **kwargs
         )
 
-    @derived_from(np.random.RandomState)
+    @derived_from(np.random.RandomState, skipblocks=1)
     def normal(self, loc=0.0, scale=1.0, size=None, chunks="auto", **kwargs):
         return self._wrap("normal", loc, scale, size=size, chunks=chunks, **kwargs)
 
-    @derived_from(np.random.RandomState)
+    @derived_from(np.random.RandomState, skipblocks=1)
     def pareto(self, a, size=None, chunks="auto", **kwargs):
         return self._wrap("pareto", a, size=size, chunks=chunks, **kwargs)
 
-    @derived_from(np.random.RandomState)
+    @derived_from(np.random.RandomState, skipblocks=1)
     def permutation(self, x):
         from .slicing import shuffle_slice
 
@@ -373,83 +373,83 @@ class RandomState(object):
         self._numpy_state.shuffle(index)
         return shuffle_slice(x, index)
 
-    @derived_from(np.random.RandomState)
+    @derived_from(np.random.RandomState, skipblocks=1)
     def poisson(self, lam=1.0, size=None, chunks="auto", **kwargs):
         return self._wrap("poisson", lam, size=size, chunks=chunks, **kwargs)
 
-    @derived_from(np.random.RandomState)
+    @derived_from(np.random.RandomState, skipblocks=1)
     def power(self, a, size=None, chunks="auto", **kwargs):
         return self._wrap("power", a, size=size, chunks=chunks, **kwargs)
 
-    @derived_from(np.random.RandomState)
+    @derived_from(np.random.RandomState, skipblocks=1)
     def randint(self, low, high=None, size=None, chunks="auto", dtype="l", **kwargs):
         return self._wrap(
             "randint", low, high, size=size, chunks=chunks, dtype=dtype, **kwargs
         )
 
-    @derived_from(np.random.RandomState)
+    @derived_from(np.random.RandomState, skipblocks=1)
     def random_integers(self, low, high=None, size=None, chunks="auto", **kwargs):
         return self._wrap(
             "random_integers", low, high, size=size, chunks=chunks, **kwargs
         )
 
-    @derived_from(np.random.RandomState)
+    @derived_from(np.random.RandomState, skipblocks=1)
     def random_sample(self, size=None, chunks="auto", **kwargs):
         return self._wrap("random_sample", size=size, chunks=chunks, **kwargs)
 
     random = random_sample
 
-    @derived_from(np.random.RandomState)
+    @derived_from(np.random.RandomState, skipblocks=1)
     def rayleigh(self, scale=1.0, size=None, chunks="auto", **kwargs):
         return self._wrap("rayleigh", scale, size=size, chunks=chunks, **kwargs)
 
-    @derived_from(np.random.RandomState)
+    @derived_from(np.random.RandomState, skipblocks=1)
     def standard_cauchy(self, size=None, chunks="auto", **kwargs):
         return self._wrap("standard_cauchy", size=size, chunks=chunks, **kwargs)
 
-    @derived_from(np.random.RandomState)
+    @derived_from(np.random.RandomState, skipblocks=1)
     def standard_exponential(self, size=None, chunks="auto", **kwargs):
         return self._wrap("standard_exponential", size=size, chunks=chunks, **kwargs)
 
-    @derived_from(np.random.RandomState)
+    @derived_from(np.random.RandomState, skipblocks=1)
     def standard_gamma(self, shape, size=None, chunks="auto", **kwargs):
         return self._wrap("standard_gamma", shape, size=size, chunks=chunks, **kwargs)
 
-    @derived_from(np.random.RandomState)
+    @derived_from(np.random.RandomState, skipblocks=1)
     def standard_normal(self, size=None, chunks="auto", **kwargs):
         return self._wrap("standard_normal", size=size, chunks=chunks, **kwargs)
 
-    @derived_from(np.random.RandomState)
+    @derived_from(np.random.RandomState, skipblocks=1)
     def standard_t(self, df, size=None, chunks="auto", **kwargs):
         return self._wrap("standard_t", df, size=size, chunks=chunks, **kwargs)
 
-    @derived_from(np.random.RandomState)
+    @derived_from(np.random.RandomState, skipblocks=1)
     def tomaxint(self, size=None, chunks="auto", **kwargs):
         return self._wrap("tomaxint", size=size, chunks=chunks, **kwargs)
 
-    @derived_from(np.random.RandomState)
+    @derived_from(np.random.RandomState, skipblocks=1)
     def triangular(self, left, mode, right, size=None, chunks="auto", **kwargs):
         return self._wrap(
             "triangular", left, mode, right, size=size, chunks=chunks, **kwargs
         )
 
-    @derived_from(np.random.RandomState)
+    @derived_from(np.random.RandomState, skipblocks=1)
     def uniform(self, low=0.0, high=1.0, size=None, chunks="auto", **kwargs):
         return self._wrap("uniform", low, high, size=size, chunks=chunks, **kwargs)
 
-    @derived_from(np.random.RandomState)
+    @derived_from(np.random.RandomState, skipblocks=1)
     def vonmises(self, mu, kappa, size=None, chunks="auto", **kwargs):
         return self._wrap("vonmises", mu, kappa, size=size, chunks=chunks, **kwargs)
 
-    @derived_from(np.random.RandomState)
+    @derived_from(np.random.RandomState, skipblocks=1)
     def wald(self, mean, scale, size=None, chunks="auto", **kwargs):
         return self._wrap("wald", mean, scale, size=size, chunks=chunks, **kwargs)
 
-    @derived_from(np.random.RandomState)
+    @derived_from(np.random.RandomState, skipblocks=1)
     def weibull(self, a, size=None, chunks="auto", **kwargs):
         return self._wrap("weibull", a, size=size, chunks=chunks, **kwargs)
 
-    @derived_from(np.random.RandomState)
+    @derived_from(np.random.RandomState, skipblocks=1)
     def zipf(self, a, size=None, chunks="auto", **kwargs):
         return self._wrap("zipf", a, size=size, chunks=chunks, **kwargs)
 
diff --git a/dask/dataframe/core.py b/dask/dataframe/core.py
index 030f67136..0c5aeba50 100644
--- a/dask/dataframe/core.py
+++ b/dask/dataframe/core.py
@@ -1684,6 +1684,18 @@ Dask Name: {name}, {task} tasks"""
                 result.divisions = (self.columns.min(), self.columns.max())
             return result
 
+    @derived_from(pd.DataFrame)
+    def mode(self, dropna=True, split_every=False):
+        mode_series = self.reduction(
+            chunk=M.value_counts,
+            combine=M.sum,
+            aggregate=_mode_aggregate,
+            split_every=split_every,
+            chunk_kwargs={"dropna": dropna},
+            aggregate_kwargs={"dropna": dropna},
+        )
+        return mode_series
+
     @derived_from(pd.DataFrame)
     def mean(self, axis=None, skipna=True, split_every=False, dtype=None, out=None):
         axis = self._validate_axis(axis)
@@ -2812,6 +2824,10 @@ Dask Name: {name}, {task} tasks""".format(
     def count(self, split_every=False):
         return super(Series, self).count(split_every=split_every)
 
+    @derived_from(pd.Series)
+    def mode(self, dropna=True, split_every=False):
+        return super(Series, self).mode(dropna=dropna, split_every=split_every)
+
     @derived_from(pd.Series, version="0.25.0")
     def explode(self):
         meta = self._meta.explode()
@@ -3343,7 +3359,7 @@ class DataFrame(_Frame):
 
     def __len__(self):
         try:
-            s = self[self.columns[0]]
+            s = self.iloc[:, 0]
         except IndexError:
             return super().__len__()
         else:
@@ -4162,6 +4178,36 @@ class DataFrame(_Frame):
     def round(self, decimals=0):
         return elemwise(M.round, self, decimals)
 
+    @derived_from(pd.DataFrame)
+    def mode(self, dropna=True, split_every=False):
+        mode_series_list = []
+        for col_index in range(len(self.columns)):
+            col_series = self.iloc[:, col_index]
+            mode_series = Series.mode(
+                col_series, dropna=dropna, split_every=split_every
+            )
+            mode_series.name = col_series.name
+            mode_series_list.append(mode_series)
+
+        name = "concat-" + tokenize(*mode_series_list)
+
+        dsk = {
+            (name, 0): (
+                apply,
+                methods.concat,
+                [[(df._name, 0) for df in mode_series_list]],
+                {"axis": 1},
+            )
+        }
+
+        meta = methods.concat([df._meta for df in mode_series_list], axis=1)
+        graph = HighLevelGraph.from_collections(
+            name, dsk, dependencies=mode_series_list
+        )
+        ddf = new_dd_object(graph, name, meta, divisions=(None, None))
+
+        return ddf
+
     @derived_from(pd.DataFrame)
     def cov(self, min_periods=None, split_every=False):
         return cov_corr(self, min_periods, split_every=split_every)
@@ -5964,6 +6010,18 @@ def idxmaxmin_agg(x, fn=None, skipna=True, scalar=False):
     return res
 
 
+def _mode_aggregate(df, dropna):
+    value_count_series = df.sum()
+    max_val = value_count_series.max(skipna=dropna)
+    mode_series = (
+        value_count_series[value_count_series == max_val]
+        .index.to_series()
+        .sort_values()
+        .reset_index(drop=True)
+    )
+    return mode_series
+
+
 def _count_aggregate(x):
     return x.sum().astype("int64")
 
diff --git a/dask/dataframe/io/parquet/arrow.py b/dask/dataframe/io/parquet/arrow.py
index 1b54d3579..b75a764d0 100644
--- a/dask/dataframe/io/parquet/arrow.py
+++ b/dask/dataframe/io/parquet/arrow.py
@@ -89,11 +89,24 @@ def _merge_statistics(stats, s):
         min_i = stats[-1]["columns"][i]["min"]
         max_i = stats[-1]["columns"][i]["max"]
         stats[-1]["columns"][i]["min"] = min(min_i, min_n)
-        stats[-1]["columns"][i]["max"] = min(max_i, max_n)
+        stats[-1]["columns"][i]["max"] = max(max_i, max_n)
         stats[-1]["columns"][i]["null_count"] += null_count_n
     return True
 
 
+class SimplePiece:
+    """ SimplePiece
+        Surrogate class for PyArrow ParquetDatasetPiece.
+        Only used for flat datasets (not partitioned) where
+        a "_metadata" file is available.
+    """
+
+    def __init__(self, path):
+        self.path = path
+        self.partition_keys = None
+        self.row_group = None
+
+
 def _determine_dataset_parts(fs, paths, gather_statistics, filters, dataset_kwargs):
     """ Determine how to access metadata and break read into ``parts``
 
@@ -103,33 +116,49 @@ def _determine_dataset_parts(fs, paths, gather_statistics, filters, dataset_kwar
     """
     parts = []
     if len(paths) > 1:
-        if gather_statistics is not False:
-            # This scans all the files
-            dataset = pq.ParquetDataset(
-                paths, filesystem=fs, filters=filters, **dataset_kwargs
-            )
-        else:
-            base, fns = _analyze_paths(paths, fs)
-            if "_metadata" in fns:
-                # We have a _metadata file, lets use it
+        base, fns = _analyze_paths(paths, fs)
+        if "_metadata" in fns:
+            # We have a _metadata file
+            # PyArrow cannot handle "_metadata"
+            # when `paths` is a list.
+            paths.remove(base + fs.sep + "_metadata")
+            fns.remove("_metadata")
+            if gather_statistics is not False:
+                # If we are allowed to gather statistics,
+                # lets use "_metadata" instead of opening
+                # every file. Note that we don't need to check if
+                # the dataset is flat here, because PyArrow cannot
+                # properly handle partitioning in this case anyway.
                 dataset = pq.ParquetDataset(
                     base + fs.sep + "_metadata",
                     filesystem=fs,
                     filters=filters,
                     **dataset_kwargs,
                 )
-            else:
-                # Rely on metadata for 0th file.
-                # Will need to pass a list of paths to read_partition
-                dataset = pq.ParquetDataset(paths[0], filesystem=fs, **dataset_kwargs)
-                parts = [base + fs.sep + fn for fn in fns]
+                dataset.metadata = dataset.pieces[0].get_metadata()
+                dataset.pieces = [SimplePiece(path) for path in paths]
+                dataset.partitions = None
+                return parts, dataset
+        if gather_statistics is not False:
+            # This scans all the files
+            dataset = pq.ParquetDataset(
+                paths, filesystem=fs, filters=filters, **dataset_kwargs
+            )
+        else:
+            # Rely on schema for 0th file.
+            # Will need to pass a list of paths to read_partition
+            dataset = pq.ParquetDataset(paths[0], filesystem=fs, **dataset_kwargs)
+            parts = [base + fs.sep + fn for fn in fns]
     elif fs.isdir(paths[0]):
         # This is a directory, check for _metadata, then _common_metadata
         allpaths = fs.glob(paths[0] + fs.sep + "*")
         base, fns = _analyze_paths(allpaths, fs)
+        # Check if dataset is "not flat" (partitioned into directories).
+        # If so, we will need to let pyarrow generate the `dataset` object.
+        not_flat = any([fs.isdir(p) for p in fs.glob(fs.sep.join([base, "*"]))])
         if "_metadata" in fns and "validate_schema" not in dataset_kwargs:
             dataset_kwargs["validate_schema"] = False
-        if "_metadata" in fns or gather_statistics is not False:
+        if not_flat or "_metadata" in fns or gather_statistics is not False:
             # Let arrow do its thing (use _metadata or scan files)
             dataset = pq.ParquetDataset(
                 paths, filesystem=fs, filters=filters, **dataset_kwargs
@@ -145,7 +174,7 @@ def _determine_dataset_parts(fs, paths, gather_statistics, filters, dataset_kwar
                 dataset = pq.ParquetDataset(
                     allpaths[0], filesystem=fs, **dataset_kwargs
                 )
-            parts = [base + fs.sep + fn for fn in fns]
+            parts = [base + fs.sep + fn for fn in fns if fn != "_common_metadata"]
     else:
         # There is only one file to read
         dataset = pq.ParquetDataset(paths, filesystem=fs, **dataset_kwargs)
diff --git a/dask/optimization.py b/dask/optimization.py
index 7a00ffc7e..0763bf771 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(
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):


Test patch:
diff --git a/dask/array/tests/test_creation.py b/dask/array/tests/test_creation.py
index 704f13e3d..ed27d227c 100644
--- a/dask/array/tests/test_creation.py
+++ b/dask/array/tests/test_creation.py
@@ -10,7 +10,7 @@ import dask
 import dask.array as da
 from dask.array.core import normalize_chunks
 from dask.array.utils import assert_eq, same_keys, AxisError
-from dask.array.numpy_compat import _numpy_117
+from dask.array.numpy_compat import _numpy_117, _numpy_118
 
 
 @pytest.mark.parametrize(
@@ -540,7 +540,7 @@ def test_repeat():
     x = np.random.random((10, 11, 13))
     d = da.from_array(x, chunks=(4, 5, 3))
 
-    repeats = [1, 2, 5]
+    repeats = [0, 1, 2, 5]
     axes = [-3, -2, -1, 0, 1, 2]
 
     for r in repeats:
@@ -719,6 +719,66 @@ def test_pad(shape, chunks, pad_width, mode, kwargs):
         assert_eq(np_r, da_r)
 
 
+@pytest.mark.parametrize("dtype", [np.uint8, np.int16, np.float32, bool])
+@pytest.mark.parametrize(
+    "pad_widths", [2, (2,), (2, 3), ((2, 3),), ((3, 1), (0, 0), (2, 0))]
+)
+@pytest.mark.parametrize(
+    "mode",
+    [
+        "constant",
+        "edge",
+        pytest.param(
+            "linear_ramp",
+            marks=pytest.mark.skipif(
+                not _numpy_118, reason="numpy changed pad behaviour"
+            ),
+        ),
+        "maximum",
+        pytest.param(
+            "mean",
+            marks=pytest.mark.skip(
+                reason="Bug dask changes the dtype to float: https://github.com/dask/dask/issues/5303"
+            ),
+        ),
+        "minimum",
+        pytest.param(
+            "reflect",
+            marks=pytest.mark.skip(
+                reason="Bug when pad_width is larger than dimension: https://github.com/dask/dask/issues/5303"
+            ),
+        ),
+        pytest.param(
+            "symmetric",
+            marks=pytest.mark.skip(
+                reason="Bug when pad_width is larger than dimension: https://github.com/dask/dask/issues/5303"
+            ),
+        ),
+        pytest.param(
+            "wrap",
+            marks=pytest.mark.skip(
+                reason="Bug when pad_width is larger than dimension: https://github.com/dask/dask/issues/5303"
+            ),
+        ),
+        pytest.param("median", marks=pytest.mark.skip(reason="Not implemented"),),
+        pytest.param(
+            "empty",
+            marks=pytest.mark.skip(
+                reason="Empty leads to undefined values, which may be different"
+            ),
+        ),
+    ],
+)
+def test_pad_3d_data(dtype, pad_widths, mode):
+    np_a = np.arange(2 * 3 * 4).reshape(2, 3, 4).astype(dtype)
+    da_a = da.from_array(np_a, chunks="auto")
+
+    np_r = np.pad(np_a, pad_widths, mode=mode)
+    da_r = da.pad(da_a, pad_widths, mode=mode)
+
+    assert_eq(np_r, da_r)
+
+
 @pytest.mark.parametrize("kwargs", [{}, {"scaler": 2}])
 def test_pad_udf(kwargs):
     def udf_pad(vector, pad_width, iaxis, kwargs):
diff --git a/dask/dataframe/io/tests/test_csv.py b/dask/dataframe/io/tests/test_csv.py
index d055458d2..a41e27ea1 100644
--- a/dask/dataframe/io/tests/test_csv.py
+++ b/dask/dataframe/io/tests/test_csv.py
@@ -1363,6 +1363,21 @@ def test_to_csv_gzip():
             tm.assert_frame_equal(result, df)
 
 
+def test_to_csv_nodir():
+    # See #6062 https://github.com/intake/filesystem_spec/pull/271 and
+    df0 = pd.DataFrame(
+        {"x": ["a", "b", "c", "d"], "y": [1, 2, 3, 4]}, index=[1.0, 2.0, 3.0, 4.0]
+    )
+    df = dd.from_pandas(df0, npartitions=2)
+    with tmpdir() as dir:
+        dir0 = os.path.join(str(dir), "createme")
+        df.to_csv(dir0)
+        assert "createme" in os.listdir(dir)
+        assert os.listdir(dir0)
+        result = dd.read_csv(os.path.join(dir0, "*")).compute()
+    assert (result.x.values == df0.x.values).all()
+
+
 def test_to_csv_simple():
     df0 = pd.DataFrame(
         {"x": ["a", "b", "c", "d"], "y": [1, 2, 3, 4]}, index=[1.0, 2.0, 3.0, 4.0]
diff --git a/dask/dataframe/io/tests/test_parquet.py b/dask/dataframe/io/tests/test_parquet.py
index 4e9361d99..92bd1284d 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)
 
 
@@ -664,6 +669,23 @@ def test_partition_on_cats(tmpdir, engine):
     assert set(df.b.cat.categories) == {"x", "y", "z"}
 
 
+@pytest.mark.parametrize("meta", [False, True])
+@pytest.mark.parametrize("stats", [False, True])
+def test_partition_on_cats_pyarrow(tmpdir, stats, meta):
+    tmp = str(tmpdir)
+    d = pd.DataFrame(
+        {
+            "a": np.random.rand(50),
+            "b": np.random.choice(["x", "y", "z"], size=50),
+            "c": np.random.choice(["x", "y", "z"], size=50),
+        }
+    )
+    d = dd.from_pandas(d, 2)
+    d.to_parquet(tmp, partition_on=["b"], engine="pyarrow", write_metadata_file=meta)
+    df = dd.read_parquet(tmp, engine="pyarrow", gather_statistics=stats)
+    assert set(df.b.cat.categories) == {"x", "y", "z"}
+
+
 def test_partition_on_cats_2(tmpdir, engine):
     tmp = str(tmpdir)
     d = pd.DataFrame(
@@ -1910,7 +1932,7 @@ def test_categories_large(tmpdir, engine):
 
 
 @write_read_engines()
-def test_read_glob_nostats(tmpdir, write_engine, read_engine):
+def test_read_glob_no_stats(tmpdir, write_engine, read_engine):
     tmp_path = str(tmpdir)
     ddf.to_parquet(tmp_path, engine=write_engine)
 
@@ -1920,6 +1942,18 @@ def test_read_glob_nostats(tmpdir, write_engine, read_engine):
     assert_eq(ddf, ddf2, check_divisions=False)
 
 
+@write_read_engines()
+def test_read_glob_yes_stats(tmpdir, write_engine, read_engine):
+    tmp_path = str(tmpdir)
+    ddf.to_parquet(tmp_path, engine=write_engine)
+    import glob
+
+    paths = glob.glob(os.path.join(tmp_path, "*.parquet"))
+    paths.append(os.path.join(tmp_path, "_metadata"))
+    ddf2 = dd.read_parquet(paths, engine=read_engine, gather_statistics=False)
+    assert_eq(ddf, ddf2, check_divisions=False)
+
+
 @pytest.mark.parametrize("statistics", [True, False, None])
 @pytest.mark.parametrize("remove_common", [True, False])
 @write_read_engines()
diff --git a/dask/dataframe/tests/test_dataframe.py b/dask/dataframe/tests/test_dataframe.py
index 5f2feddee..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():
@@ -3676,6 +3678,22 @@ def test_dataframe_reductions_arithmetic(reduction):
     )
 
 
+def test_dataframe_mode():
+    data = [["Tom", 10, 7], ["Farahn", 14, 7], ["Julie", 14, 5], ["Nick", 10, 10]]
+
+    df = pd.DataFrame(data, columns=["Name", "Num", "Num"])
+    ddf = dd.from_pandas(df, npartitions=3)
+
+    assert_eq(ddf.mode(), df.mode())
+    assert_eq(ddf.Name.mode(), df.Name.mode())
+
+    # test empty
+    df = pd.DataFrame(columns=["a", "b"])
+    ddf = dd.from_pandas(df, npartitions=1)
+    # check_index=False should be removed once https://github.com/pandas-dev/pandas/issues/33321 is resolved.
+    assert_eq(ddf.mode(), df.mode(), check_index=False)
+
+
 def test_datetime_loc_open_slicing():
     dtRange = pd.date_range("01.01.2015", "05.05.2015")
     df = pd.DataFrame(np.random.random((len(dtRange), 2)), index=dtRange)
diff --git a/dask/tests/test_optimization.py b/dask/tests/test_optimization.py
index 99e32635f..c92da2495 100644
--- a/dask/tests/test_optimization.py
+++ b/dask/tests/test_optimization.py
@@ -1297,3 +1297,39 @@ def test_dont_fuse_numpy_arrays():
     dsk = {"x": np.arange(5), "y": (inc, "x")}
 
     assert fuse(dsk, "y")[0] == dsk
+
+
+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