Repository: dask/dask

Issue:
enforce_type slowdown in meta computation in dask array
We have a LBFGS implementation where we are seeing a 40x increase in runtime during graph construction when using dask 2.0.0 compared to dask 1.2.2.

I have profiled it a bit and it seems all the additional time is spent resolving dtypes for the multiplications. I have not yet managed to reduce it to a smaller reproducer than our H^-1 * G function.

The attached code needs about 0.08 seconds per call to `calculate_hinvg` for 1.2.2 and about 3.4s for 2.0.0.

```python
import time

import numpy as np
import dask.array as da
import dask

M = 5
N= 10000
grad = da.zeros((N,), dtype=np.float64, chunks=N)
w = da.zeros((N,), dtype=np.float64, chunks=N)
search_steps = [da.zeros((N,), dtype=np.float64, chunks=N) for i in range(M)]
gradient_differences = [da.zeros((N,), dtype=np.float64, chunks=N) for i in range(M)]
mask = da.zeros((M,), dtype=np.float64, chunks=N)
rho = da.zeros((M,), dtype=np.float64, chunks=N)
stp = da.from_array([1], chunks=1)

def calculate_hinvg(grad, w, search_steps, gradient_differences, mask, stp, rho):
    """ Algorithm follows the explanation at
    https://en.wikipedia.org/wiki/Limited-memory_BFGS"""

    # For us the newest entry in the arrays is always at index 0

    # calculate the missing gradient difference and the missing search direction
    # search direction is the value that we calculated the last time multiplied
    # with the accepted step length
    grad_diff_new = [grad - w]
    grad_diff_new.extend(gradient_differences)
    search_steps_new = [stp * search_steps[0]]

    # don't add the entry that is the unscaled search direction
    search_steps_new.extend(search_steps[1:])

    ys = da.dot(grad_diff_new[0], search_steps_new[0])
    yy = da.dot(grad_diff_new[0], grad_diff_new[0])

    gamma = ys / yy
    rho = [1.0 / ys]

    work_array = -1 * grad
    for i in range(M - 1):
        rho.append(rho[i])

    alpha_list = []
    for i in range(0, M):
        sq = da.dot(search_steps_new[i], work_array)
        alpha = sq * rho[i]
        alpha_list.append(alpha)
        work_array = work_array - mask[i] * alpha * grad_diff_new[i]

    z = gamma * work_array

    for j in range(0, M):
        i = M - 1 - j
        yr = da.dot(grad_diff_new[i], z)
        beta = alpha_list[i] - rho[i] * yr
        z = z + mask[i] * beta * search_steps_new[i]

    # z is the proposed new search direction if we accept the iteration
    return z, rho, search_steps_new

for i in range(20):
    start = time.clock()
    w, rho, search_steps = calculate_hinvg(grad, w, search_steps, gradient_differences, mask, stp, rho)
    end = time.clock()
    print("Runtime: %f" % (end - start))
```

Below also a part of the cprofile that shows that the problem might be recursive resolving of the return type.

![dask2](https://user-images.githubusercontent.com/6321320/60490338-ec218700-9cd8-11e9-9237-1a69b7fb3ec4.png)


Discussion:
pentschev: Also in here there's a huge amount of failures, apparently they're coming from dask dataframe. Are you aware of those @TomAugspurger ?
TomAugspurger: I see those same failures locally on your branch, and not on master, so it seems released to your changes.

We're passing in `meta=2`  here

```
1451        def _reduction_agg(self, name, axis=None, skipna=True, split_every=False, out=None):
1452            axis = self._validate_axis(axis)
1453
1454            meta = getattr(self._meta_nonempty, name)(axis=axis, skipna=skipna)
1455            token = self._token_prefix + name
1456
1457            method = getattr(M, name)
1458            if axis == 1:
1459                result = self.map_partitions(
1460                    method, meta=meta, token=token, skipna=skipna, axis=axis
1461                )
1462                return handle_out(out, result)
1463            else:
1464                result = self.reduction(
1465                    method,
1466                    meta=meta,
1467                    token=token,
1468                    skipna=skipna,
1469                    axis=axis,
1470 ->                 split_every=split_every,
1471                )
1472                if isinstance(self, DataFrame):
1473                    result.divisions = (min(self.columns), max(self.columns))
1474                return handle_out(out, result)
(Pdb) meta
2
```

not sure if that's correct, or if it should be `int`.
pentschev: Indeed, I see now. Sorry for making you diving into this unnecessarily @TomAugspurger. I will see how can I fix this properly.
pentschev: I pushed some new changes that should address the issues we've seen. It doesn't feel like a very nice solution, since I had to split it into a broad `is_arraylike` and another to check for `Dask.Array._meta` compatible function `is_array_meta_like`. I can't think of any much better ways to handle this at the moment, if anyone can think of one, please let me know!
pentschev: Now it looks like a CI issue, could you confirm @TomAugspurger?
jrbourbeau: Could you try merging `master` into this branch, Travis isn't picking up the `ENV_FILE` environment variable added in #5038
pentschev: Done now, I probably missed the PR that fixed this in the meantime. Thanks @jrbourbeau!
TomAugspurger: I've seen the appveyor failure at https://ci.appveyor.com/project/daskdev/dask/builds/25699854#L649 elsewhere. Still debugging. Safe to ignore for now.
pentschev: Thanks for letting me know @TomAugspurger, I was just waiting for Travis to see what other failures there will be and debug all at once. :)

Patch:
diff --git a/dask/array/core.py b/dask/array/core.py
index f8db00625..4b80439ce 100644
--- a/dask/array/core.py
+++ b/dask/array/core.py
@@ -27,7 +27,7 @@ from toolz import map, reduce, frequencies
 import numpy as np
 
 from . import chunk
-from .. import config
+from .. import config, compute
 from ..base import (
     DaskMethodsMixin,
     tokenize,
@@ -138,6 +138,33 @@ def getter_inline(a, b, asarray=True, lock=None):
 from .optimization import optimize, fuse_slice
 
 
+# __array_function__ dict for mapping aliases and mismatching names
+_HANDLED_FUNCTIONS = {}
+
+
+def implements(*numpy_functions):
+    """Register an __array_function__ implementation for dask.array.Array
+
+    Register that a function implements the API of a NumPy function (or several
+    NumPy functions in case of aliases) which is handled with
+    ``__array_function__``.
+
+    Parameters
+    ----------
+    \\*numpy_functions : callables
+        One or more NumPy functions that are handled by ``__array_function__``
+        and will be mapped by `implements` to a `dask.array` function.
+    """
+
+    def decorator(dask_func):
+        for numpy_function in numpy_functions:
+            _HANDLED_FUNCTIONS[numpy_function] = dask_func
+
+        return dask_func
+
+    return decorator
+
+
 def slices_from_chunks(chunks):
     """ Translate chunks tuple to a set of slices in product order
 
@@ -1221,16 +1248,41 @@ class Array(DaskMethodsMixin):
     def __array_function__(self, func, types, args, kwargs):
         import dask.array as module
 
+        def handle_nonmatching_names(func, args, kwargs):
+            if func not in _HANDLED_FUNCTIONS:
+                warnings.warn(
+                    "The `{}` function is not implemented by Dask array. "
+                    "You may want to use the da.map_blocks function "
+                    "or something similar to silence this warning. "
+                    "Your code may stop working in a future release.".format(
+                        func.__module__ + "." + func.__name__
+                    ),
+                    FutureWarning,
+                )
+                # Need to convert to array object (e.g. numpy.ndarray or
+                # cupy.ndarray) as needed, so we can call the NumPy function
+                # again and it gets the chance to dispatch to the right
+                # implementation.
+                args, kwargs = compute(args, kwargs)
+                return func(*args, **kwargs)
+
+            return _HANDLED_FUNCTIONS[func](*args, **kwargs)
+
+        # First try to find a matching function name.  If that doesn't work, we may
+        # be dealing with an alias or a function that's simply not in the Dask API.
+        # Handle aliases via the _HANDLED_FUNCTIONS dict mapping, and warn otherwise.
         for submodule in func.__module__.split(".")[1:]:
             try:
                 module = getattr(module, submodule)
             except AttributeError:
-                return NotImplemented
+                return handle_nonmatching_names(func, args, kwargs)
+
         if not hasattr(module, func.__name__):
-            return NotImplemented
+            return handle_nonmatching_names(func, args, kwargs)
+
         da_func = getattr(module, func.__name__)
         if da_func is func:
-            return NotImplemented
+            return handle_nonmatching_names(func, args, kwargs)
         return da_func(*args, **kwargs)
 
     @property
diff --git a/dask/array/linalg.py b/dask/array/linalg.py
index 87711d540..16a4615db 100644
--- a/dask/array/linalg.py
+++ b/dask/array/linalg.py
@@ -6,7 +6,7 @@ from numbers import Number
 import numpy as np
 import toolz
 
-from ..base import tokenize
+from ..base import tokenize, wait
 from ..blockwise import blockwise
 from ..compatibility import apply
 from ..highlevelgraph import HighLevelGraph
@@ -77,7 +77,7 @@ def tsqr(data, compute_svd=False, _max_vchunk_size=None):
     Notes
     -----
     With ``k`` blocks of size ``(m, n)``, this algorithm has memory use that
-    scales as ``k * m * n``.
+    scales as ``k * n * n``.
 
     The implementation here is the recursive variant due to the ultimate
     need for one "single core" QR decomposition. In the non-recursive version
@@ -93,8 +93,9 @@ def tsqr(data, compute_svd=False, _max_vchunk_size=None):
     Where blocks are irregular, the above logic is applied with the "height" of
     the "tallest" block used in place of ``m``.
 
-    Consider use of the ``rechunk`` method to control this behavior. Blocks
-    that are as tall as possible are recommended.
+    Consider use of the ``rechunk`` method to control this behavior.
+    Taller blocks will reduce overall memory use (assuming that many of them
+    still fit in memory at once).
 
     See Also
     --------
@@ -230,7 +231,7 @@ def tsqr(data, compute_svd=False, _max_vchunk_size=None):
         dependencies = toolz.merge(q_inner.dask.dependencies, r_inner.dask.dependencies)
 
         # Q_inner: "unstack"
-        name_q_st2 = "getitem-" + token + "-q2"
+        name_q_st2 = "getitem" + token + "-q2"
         dsk_q_st2 = dict(
             (
                 (name_q_st2, j, 0),
@@ -250,13 +251,13 @@ def tsqr(data, compute_svd=False, _max_vchunk_size=None):
         dependencies[name_q_st2] = q_inner.__dask_layers__()
 
         # R: R_inner
-        name_r_st2 = "r-inner-" + token
+        name_r_st2 = "r-inner" + token
         dsk_r_st2 = {(name_r_st2, 0, 0): (r_inner.name, 0, 0)}
         layers[name_r_st2] = dsk_r_st2
         dependencies[name_r_st2] = r_inner.__dask_layers__()
 
         # Q: Block qr[0] (*) Q_inner
-        name_q_st3 = "dot-" + token + "-q3"
+        name_q_st3 = "dot" + token + "-q3"
         dsk_q_st3 = blockwise(
             np.dot,
             name_q_st3,
@@ -268,7 +269,7 @@ def tsqr(data, compute_svd=False, _max_vchunk_size=None):
             numblocks={name_q_st1: numblocks, name_q_st2: numblocks},
         )
         layers[name_q_st3] = dsk_q_st3
-        dependencies[name_q_st3] = {name_q_st1, name_q_st2, name_q_st3}
+        dependencies[name_q_st3] = {name_q_st1, name_q_st2}
     else:
         # Do single core computation
 
@@ -625,7 +626,7 @@ def compression_level(n, q, oversampling=10, min_subspace_size=20):
     return min(max(min_subspace_size, q + oversampling), n)
 
 
-def compression_matrix(data, q, n_power_iter=0, seed=None):
+def compression_matrix(data, q, n_power_iter=0, seed=None, compute=False):
     """ Randomly sample matrix to find most active subspace
 
     This compression matrix returned by this algorithm can be used to
@@ -641,6 +642,12 @@ def compression_matrix(data, q, n_power_iter=0, seed=None):
     n_power_iter: int
         number of power iterations, useful when the singular values of
         the input matrix decay very slowly.
+    compute : bool
+        Whether or not to compute data at each use.
+        Recomputing the input while performing several passes reduces memory
+        pressure, but means that we have to compute the input multiple times.
+        This is a good choice if the data is larger than memory and cheap to
+        recreate.
 
     References
     ----------
@@ -653,18 +660,28 @@ def compression_matrix(data, q, n_power_iter=0, seed=None):
     """
     n = data.shape[1]
     comp_level = compression_level(n, q)
-    state = RandomState(seed)
+    if isinstance(seed, RandomState):
+        state = seed
+    else:
+        state = RandomState(seed)
     omega = state.standard_normal(
         size=(n, comp_level), chunks=(data.chunks[1], (comp_level,))
     )
     mat_h = data.dot(omega)
     for j in range(n_power_iter):
-        mat_h = data.dot(data.T.dot(mat_h))
+        if compute:
+            mat_h = mat_h.persist()
+            wait(mat_h)
+        tmp = data.T.dot(mat_h)
+        if compute:
+            tmp = tmp.persist()
+            wait(tmp)
+        mat_h = data.dot(tmp)
     q, _ = tsqr(mat_h)
     return q.T
 
 
-def svd_compressed(a, k, n_power_iter=0, seed=None):
+def svd_compressed(a, k, n_power_iter=0, seed=None, compute=False):
     """ Randomly compressed rank-k thin Singular Value Decomposition.
 
     This computes the approximate singular value decomposition of a large
@@ -682,10 +699,15 @@ def svd_compressed(a, k, n_power_iter=0, seed=None):
         Number of power iterations, useful when the singular values
         decay slowly. Error decreases exponentially as n_power_iter
         increases. In practice, set n_power_iter <= 4.
+    compute : bool
+        Whether or not to compute data at each use.
+        Recomputing the input while performing several passes reduces memory
+        pressure, but means that we have to compute the input multiple times.
+        This is a good choice if the data is larger than memory and cheap to
+        recreate.
 
     Examples
     --------
-
     >>> u, s, vt = svd_compressed(x, 20)  # doctest: +SKIP
 
     Returns
@@ -703,7 +725,12 @@ def svd_compressed(a, k, n_power_iter=0, seed=None):
     pp. 217-288, June 2011
     https://arxiv.org/abs/0909.4061
     """
-    comp = compression_matrix(a, k, n_power_iter=n_power_iter, seed=seed)
+    comp = compression_matrix(
+        a, k, n_power_iter=n_power_iter, seed=seed, compute=compute
+    )
+    if compute:
+        comp = comp.persist()
+        wait(comp)
     a_compressed = comp.dot(a)
     v, s, u = tsqr(a_compressed.T, compute_svd=True)
     u = comp.T.dot(u)
diff --git a/dask/array/reductions.py b/dask/array/reductions.py
index 9c7cc1921..beb42fd51 100644
--- a/dask/array/reductions.py
+++ b/dask/array/reductions.py
@@ -11,7 +11,7 @@ from numbers import Integral, Number
 from toolz import compose, partition_all, get, accumulate, pluck
 
 from . import chunk
-from .core import _concatenate2, Array, handle_out
+from .core import _concatenate2, Array, handle_out, implements
 from .blockwise import blockwise
 from ..blockwise import lol_tuples
 from .creation import arange, diagonal
@@ -363,6 +363,7 @@ def prod(a, axis=None, dtype=None, keepdims=False, split_every=None, out=None):
     )
 
 
+@implements(np.min, np.amin)
 @wraps(chunk.min)
 def min(a, axis=None, keepdims=False, split_every=None, out=None):
     return reduction(
@@ -377,6 +378,7 @@ def min(a, axis=None, keepdims=False, split_every=None, out=None):
     )
 
 
+@implements(np.max, np.amax)
 @wraps(chunk.max)
 def max(a, axis=None, keepdims=False, split_every=None, out=None):
     return reduction(
diff --git a/dask/array/routines.py b/dask/array/routines.py
index de3606ddd..5f7b3a754 100644
--- a/dask/array/routines.py
+++ b/dask/array/routines.py
@@ -35,6 +35,7 @@ from .core import (
     is_scalar_for_elemwise,
     broadcast_to,
     tensordot_lookup,
+    implements,
 )
 
 from .einsumfuncs import einsum  # noqa
@@ -804,6 +805,7 @@ def corrcoef(x, y=None, rowvar=1):
     return (c / sqr_d) / sqr_d.T
 
 
+@implements(np.round, np.round_)
 @derived_from(np)
 def round(a, decimals=0):
     return a.map_blocks(np.round, decimals=decimals, dtype=a.dtype)
diff --git a/dask/array/svg.py b/dask/array/svg.py
index d59d1df9b..d681bd1d9 100644
--- a/dask/array/svg.py
+++ b/dask/array/svg.py
@@ -23,6 +23,8 @@ def svg(chunks, size=200, **kwargs):
     if np.isnan(shape).any():  # don't support unknown sizes
         raise NotImplementedError("Can't generate SVG with unknown chunk sizes")
     if not all(shape):
+        raise NotImplementedError("Can't generate SVG with 0-length dimensions")
+    if len(chunks) == 0:
         raise NotImplementedError("Can't generate SVG with 0 dimensions")
     if len(chunks) == 1:
         return svg_1d(chunks, size=size, **kwargs)
diff --git a/dask/base.py b/dask/base.py
index 8f1f7915e..1be5b864c 100644
--- a/dask/base.py
+++ b/dask/base.py
@@ -1011,3 +1011,18 @@ def get_scheduler(get=None, scheduler=None, collections=None, cls=None):
         return get
 
     return None
+
+
+def wait(x, timeout=None, return_when="ALL_COMPLETED"):
+    """ Wait until computation has finished
+
+    This is a compatibility alias for ``dask.distributed.wait``.
+    If it is applied onto Dask collections without Dask Futures or if Dask
+    distributed is not installed then it is a no-op
+    """
+    try:
+        from distributed import wait
+
+        return wait(x, timeout=timeout, return_when=return_when)
+    except (ImportError, ValueError):
+        return x
diff --git a/dask/dataframe/core.py b/dask/dataframe/core.py
index d2a80aac6..a19559e09 100644
--- a/dask/dataframe/core.py
+++ b/dask/dataframe/core.py
@@ -4759,7 +4759,7 @@ def map_partitions(func, *args, **kwargs):
         }
         graph = HighLevelGraph.from_collections(name, layer, dependencies=args)
         return Scalar(graph, name, meta)
-    elif not (has_parallel_type(meta) or is_arraylike(meta)):
+    elif not (has_parallel_type(meta) or is_arraylike(meta) and meta.shape):
         # If `meta` is not a pandas object, the concatenated results will be a
         # different type
         meta = make_meta(_concat([meta]), index=meta_index)
@@ -5841,7 +5841,7 @@ def new_dd_object(dsk, name, meta, divisions):
     """
     if has_parallel_type(meta):
         return get_parallel_type(meta)(dsk, name, meta, divisions)
-    elif is_arraylike(meta):
+    elif is_arraylike(meta) and meta.shape:
         import dask.array as da
 
         chunks = ((np.nan,) * (len(divisions) - 1),) + tuple(
diff --git a/dask/dataframe/reshape.py b/dask/dataframe/reshape.py
index bb62c38e2..8377b40d6 100644
--- a/dask/dataframe/reshape.py
+++ b/dask/dataframe/reshape.py
@@ -7,7 +7,7 @@ from .core import Series, DataFrame, map_partitions, apply_concat_apply
 from . import methods
 from .utils import is_categorical_dtype, is_scalar, has_known_categories, PANDAS_VERSION
 from ..utils import M
-
+import sys
 
 ###############################################################
 # Dummies
@@ -23,6 +23,7 @@ def get_dummies(
     sparse=False,
     drop_first=False,
     dtype=np.uint8,
+    **kwargs
 ):
     """
     Convert categorical variable into dummy/indicator variables.
@@ -106,7 +107,7 @@ def get_dummies(
     """
     if PANDAS_VERSION >= "0.23.0":
         # dtype added to pandas
-        kwargs = {"dtype": dtype}
+        kwargs["dtype"] = dtype
     elif dtype != np.uint8:
         # User specified something other than the default.
         raise ValueError(
@@ -114,8 +115,6 @@ def get_dummies(
             "The 'dtype' keyword was added in pandas "
             "0.23.0.".format(PANDAS_VERSION)
         )
-    else:
-        kwargs = {}
 
     if isinstance(data, (pd.Series, pd.DataFrame)):
         return pd.get_dummies(
@@ -161,7 +160,9 @@ def get_dummies(
 
     # We explicitly create `meta` on `data._meta` (the empty version) to
     # work around https://github.com/pandas-dev/pandas/issues/21993
-    meta = pd.get_dummies(
+    package_name = data._meta.__class__.__module__.split(".")[0]
+    dummies = sys.modules[package_name].get_dummies
+    meta = dummies(
         data._meta,
         prefix=prefix,
         prefix_sep=prefix_sep,
@@ -173,7 +174,7 @@ def get_dummies(
     )
 
     return map_partitions(
-        pd.get_dummies,
+        dummies,
         data,
         prefix=prefix,
         prefix_sep=prefix_sep,
diff --git a/dask/dataframe/utils.py b/dask/dataframe/utils.py
index 04c56cb76..bf11428d1 100644
--- a/dask/dataframe/utils.py
+++ b/dask/dataframe/utils.py
@@ -317,7 +317,7 @@ def make_meta_object(x, index=None):
     """
     if hasattr(x, "_meta"):
         return x._meta
-    elif is_arraylike(x):
+    elif is_arraylike(x) and x.shape:
         return x[:0]
 
     if index is not None:
diff --git a/dask/utils.py b/dask/utils.py
index 74638e090..532fe30c2 100644
--- a/dask/utils.py
+++ b/dask/utils.py
@@ -1051,8 +1051,9 @@ def is_arraylike(x):
     Examples
     --------
     >>> import numpy as np
-    >>> x = np.ones(5)
-    >>> is_arraylike(x)
+    >>> is_arraylike(np.ones(5))
+    True
+    >>> is_arraylike(np.ones(()))
     True
     >>> is_arraylike(5)
     False
@@ -1061,10 +1062,9 @@ def is_arraylike(x):
     """
     from .base import is_dask_collection
 
-    return (
+    return bool(
         hasattr(x, "shape")
         and isinstance(x.shape, tuple)
-        and x.shape
         and hasattr(x, "dtype")
         and not any(is_dask_collection(n) for n in x.shape)
     )
diff --git a/docs/source/delayed.rst b/docs/source/delayed.rst
index f35a50418..54b85c9a5 100644
--- a/docs/source/delayed.rst
+++ b/docs/source/delayed.rst
@@ -21,7 +21,7 @@ directly with a light annotation of normal python code:
    >>> z = dask.delayed(add)(x, y)
    >>> z.compute()
    5
-   >>> z.vizualize()
+   >>> z.visualize()
 
 .. image:: images/inc-add.svg
    :alt: simple task graph created with dask.delayed
diff --git a/docs/source/setup/kubernetes-helm.rst b/docs/source/setup/kubernetes-helm.rst
index aa2cf7f95..91972f6f8 100644
--- a/docs/source/setup/kubernetes-helm.rst
+++ b/docs/source/setup/kubernetes-helm.rst
@@ -28,8 +28,8 @@ Microsoft.  We recommend the first part of the documentation in the guide
 that focuses on Kubernetes and Helm (you do not need to follow all of these
 instructions).  Also, JupyterHub is not necessary to deploy Dask:
 
-- `Creating a Kubernetes Cluster <https://zero-to-jupyterhub.readthedocs.io/en/v0.4-doc/create-k8s-cluster.html>`_
-- `Setting up Helm <https://zero-to-jupyterhub.readthedocs.io/en/v0.4-doc/setup-helm.html>`_
+- `Creating a Kubernetes Cluster <https://zero-to-jupyterhub.readthedocs.io/en/latest/create-k8s-cluster.html>`_
+- `Setting up Helm <https://zero-to-jupyterhub.readthedocs.io/en/latest/setup-helm.html>`_
 
 Alternatively, you may want to experiment with Kubernetes locally using
 `Minikube <https://kubernetes.io/docs/getting-started-guides/minikube/>`_.


Test patch:
diff --git a/dask/array/tests/test_array_function.py b/dask/array/tests/test_array_function.py
index 37d1779bc..17ee8d51d 100644
--- a/dask/array/tests/test_array_function.py
+++ b/dask/array/tests/test_array_function.py
@@ -21,10 +21,14 @@ missing_arrfunc_reason = "NEP-18 support is not available in NumPy"
         lambda x: np.matmul(x, x),
         lambda x: np.mean(x),
         lambda x: np.stack([x, x]),
+        lambda x: np.block([x, x]),
         lambda x: np.sum(x),
         lambda x: np.var(x),
         lambda x: np.vstack(x),
         lambda x: np.linalg.norm(x),
+        lambda x: np.min(x),
+        lambda x: np.amin(x),
+        lambda x: np.round(x),
     ],
 )
 def test_array_function_dask(func):
@@ -57,13 +61,16 @@ def test_array_function_fft(func):
         lambda x: np.min_scalar_type(x),
         lambda x: np.linalg.det(x),
         lambda x: np.linalg.eigvals(x),
+        lambda x: np.median(x),
     ],
 )
 def test_array_notimpl_function_dask(func):
     x = np.random.random((100, 100))
     y = da.from_array(x, chunks=(50, 50))
 
-    with pytest.raises(TypeError):
+    with pytest.warns(
+        FutureWarning, match="The `.*` function is not implemented by Dask"
+    ):
         func(y)
 
 
@@ -220,3 +227,17 @@ def test_unregistered_func(func):
 
     # Check that they are equivalent arrays.
     assert_eq(xx, yy, check_meta=False)
+
+
+def test_median_func():
+    # Regression test for __array_function__ becoming default in numpy 1.17
+    # dask has no median function, so ensure that this still calls np.median
+    image = da.from_array(np.array([[0, 1], [1, 2]]), chunks=(1, 2))
+    if IS_NEP18_ACTIVE:
+        with pytest.warns(
+            FutureWarning,
+            match="The `numpy.median` function is not implemented by Dask",
+        ):
+            assert int(np.median(image)) == 1
+    else:
+        assert int(np.median(image)) == 1
diff --git a/dask/array/tests/test_linalg.py b/dask/array/tests/test_linalg.py
index 4ce4d6127..d53b91a8b 100644
--- a/dask/array/tests/test_linalg.py
+++ b/dask/array/tests/test_linalg.py
@@ -910,3 +910,13 @@ def test_norm_implemented_errors(shape, chunks, axis, norm, keepdims):
     if len(shape) > 2 and len(axis) == 2:
         with pytest.raises(NotImplementedError):
             da.linalg.norm(d, ord=norm, axis=axis, keepdims=keepdims)
+
+
+def test_svd_compressed_compute():
+    x = da.ones((100, 100), chunks=(10, 10))
+    u, s, v = da.linalg.svd_compressed(x, k=2, n_power_iter=0, compute=True, seed=123)
+    uu, ss, vv = da.linalg.svd_compressed(x, k=2, n_power_iter=0, seed=123)
+
+    assert len(v.dask) < len(vv.dask)
+
+    assert_eq(v, vv)
diff --git a/dask/array/tests/test_numpy_compat.py b/dask/array/tests/test_numpy_compat.py
index 14aa924b9..14d07735c 100644
--- a/dask/array/tests/test_numpy_compat.py
+++ b/dask/array/tests/test_numpy_compat.py
@@ -35,3 +35,13 @@ def test_slice_dtype(dtype, index):
     result = _make_sliced_dtype(dtype, index)
     expected = np.ones((5, len(dtype)), dtype=dtype)[index].dtype
     assert result == expected
+
+
+def test_min_max_round_funcs():
+    # Regression test for gh-5031
+    image = da.from_array(np.array([[0, 1], [1, 2]]), chunks=(1, 2))
+    # These use __array_function__ (and min/max/round are aliased,
+    # to amin/amax/round_ in numpy)
+    assert int(np.min(image)) == 0
+    assert int(np.max(image)) == 2
+    assert np.round(image)[1, 1] == 2
diff --git a/dask/array/tests/test_reductions.py b/dask/array/tests/test_reductions.py
index 3ab663334..c17e919d1 100644
--- a/dask/array/tests/test_reductions.py
+++ b/dask/array/tests/test_reductions.py
@@ -507,6 +507,7 @@ def test_reduction_names():
     assert x.mean().name.startswith("mean")
 
 
+@pytest.mark.filterwarnings("ignore:`argmax` is not implemented by dask")
 @pytest.mark.parametrize("func", [np.sum, np.argmax])
 def test_array_reduction_out(func):
     x = da.arange(10, chunks=(5,))
diff --git a/dask/array/tests/test_svg.py b/dask/array/tests/test_svg.py
index da3ed848d..4cb29bc58 100644
--- a/dask/array/tests/test_svg.py
+++ b/dask/array/tests/test_svg.py
@@ -20,6 +20,7 @@ def test_basic():
 
 
 def test_repr_html():
+    assert da.ones([])._repr_html_()
     assert da.ones(10)[:0]._repr_html_()
     assert da.ones(10)._repr_html_()
     assert da.ones((10, 10))._repr_html_()
@@ -28,13 +29,27 @@ def test_repr_html():
 
 
 def test_errors():
-    with pytest.raises(NotImplementedError):
-        assert da.ones(10)[:0].to_svg()
-
-    with pytest.raises(NotImplementedError):
+    # empty arrays
+    with pytest.raises(NotImplementedError) as excpt:
+        da.ones([]).to_svg()
+    assert "0 dimensions" in str(excpt.value)
+
+    # Scalars
+    with pytest.raises(NotImplementedError) as excpt:
+        da.asarray(1).to_svg()
+    assert "0 dimensions" in str(excpt.value)
+
+    # 0-length dims arrays
+    with pytest.raises(NotImplementedError) as excpt:
+        da.ones(10)[:0].to_svg()
+    assert "0-length dimensions" in str(excpt.value)
+
+    # unknown chunk sizes
+    with pytest.raises(NotImplementedError) as excpt:
         x = da.ones(10)
         x = x[x > 5]
         x.to_svg()
+    assert "unknown chunk sizes" in str(excpt.value)
 
 
 def test_repr_html():
diff --git a/dask/array/tests/test_ufunc.py b/dask/array/tests/test_ufunc.py
index e00b2eea0..9ee60c733 100644
--- a/dask/array/tests/test_ufunc.py
+++ b/dask/array/tests/test_ufunc.py
@@ -237,6 +237,7 @@ def test_ufunc_outer():
 def test_complex(ufunc):
 
     dafunc = getattr(da, ufunc)
+    # Note that these functions are not NumPy ufuncs
     npfunc = getattr(np, ufunc)
 
     real = np.random.randint(1, 100, size=(20, 20))
@@ -255,7 +256,6 @@ def test_complex(ufunc):
         # applying Dask ufunc doesn't trigger computation
         assert isinstance(dafunc(darr), da.Array)
         assert_eq(dafunc(darr), npfunc(arr))
-
         assert_eq(npfunc(darr), npfunc(arr))
 
         # applying Dask ufunc to normal ndarray triggers computation
@@ -394,7 +394,7 @@ def test_array_ufunc():
     x = np.arange(24).reshape((4, 6))
     d = da.from_array(x, chunks=(2, 3))
 
-    for func in [np.sin, np.isreal, np.sum, np.negative, partial(np.prod, axis=0)]:
+    for func in [np.sin, np.sum, np.negative, partial(np.prod, axis=0)]:
         assert isinstance(func(d), da.Array)
         assert_eq(func(d), func(x))
 
diff --git a/dask/tests/test_utils.py b/dask/tests/test_utils.py
index 2817e2764..3b9aece3c 100644
--- a/dask/tests/test_utils.py
+++ b/dask/tests/test_utils.py
@@ -27,6 +27,7 @@ from dask.utils import (
     derived_from,
     parse_timedelta,
     parse_bytes,
+    is_arraylike,
 )
 from dask.utils_test import inc
 from dask.highlevelgraph import HighLevelGraph
@@ -506,3 +507,15 @@ def test_parse_timedelta():
     assert parse_timedelta("1", default="seconds") == 1
     assert parse_timedelta("1", default="ms") == 0.001
     assert parse_timedelta(1, default="ms") == 0.001
+
+
+def test_is_arraylike():
+    assert is_arraylike(0) is False
+    assert is_arraylike(()) is False
+    assert is_arraylike((0)) is False
+    assert is_arraylike([]) is False
+    assert is_arraylike([0]) is False
+
+    assert is_arraylike(np.empty(())) is True
+    assert is_arraylike(np.empty((0,))) is True
+    assert is_arraylike(np.empty((0, 0))) is True