Repository: dask/dask

Issue:
numpy.broadcast_array on dask array raises ValueError("Chunks do not align").
This is a regression. The old behaviour seems to be broadcast_array forces dask to perform a compute() to return arrays instead.

The new current behaviour seems to be going through the `__array_function__` protocol and perform a dask broadcast without automatic rechunking.

I am not sure when this has changed; and I don't know if there is anything we can do about this on the dask side -- eliminating implicit calls to compute() will surely cause some minor pains like this. 

An implicit rechunking is probably appropriate in this context, but I worry the operation may not always be well-defined (I couldn't think of an example where it is not.)

The stack trace
```
  File "<__array_function__ internals>", line 6, in broadcast_arrays
  File "/dev/shm/local/lib/python3.7/site-packages/dask/array/core.py", line 1359, in __array_function__
    return da_func(*args, **kwargs)
  File "/dev/shm/local/lib/python3.7/site-packages/dask/array/core.py", line 4025, in broadcast_arrays
    chunks = broadcast_chunks(*(e.chunks for e in args))
  File "/dev/shm/local/lib/python3.7/site-packages/dask/array/core.py", line 779, in broadcast_chunks
    raise ValueError("Chunks do not align: %s" % str(step2))
ValueError: Chunks do not align: [(55276, 100000, 100000, 100000, 100000, 100000, 100000, 100000, 100000, 100000, 100000, 100000, 100000, 100000, 100000, 100000, 100000, 100000, 100000, 100000, 100000, 100000, 100000, 100000, 100000, 100000, 100000, 100000, 100000, 100000, 100000, 100000, 100000, 100000, 100000, 100000, 100000, 100000, 100000, 100000, 100000, 100000, 100000, 100000, 100000, 100000, 100000, 100000, 100000, 100000, 100000, 100000, 100000, 100000, 100000, 100000, 100000, 100000, 100000, 100000, 100000, 100000, 100000, 100000, 100000, 100000, 100000, 100000, 100000, 100000, 100000, 100000, 100000, 100000, 100000, 100000, 100000, 100000, 100000, 100000, 100000, 100000, 100000, 100000, 100000, 100000, 13022), (8568298,)]
```



Discussion:

Patch:
diff --git a/dask/array/core.py b/dask/array/core.py
index 75b7682de..3938fe3fe 100644
--- a/dask/array/core.py
+++ b/dask/array/core.py
@@ -4025,6 +4025,11 @@ def broadcast_arrays(*args, **kwargs):
     if kwargs:
         raise TypeError("unsupported keyword argument(s) provided")
 
+    # Unify uneven chunking
+    inds = [list(reversed(range(x.ndim))) for x in args]
+    uc_args = concat(zip(args, inds))
+    _, args = unify_chunks(*uc_args, warn=False)
+
     shape = broadcast_shapes(*(e.shape for e in args))
     chunks = broadcast_chunks(*(e.chunks for e in args))
 
diff --git a/dask/bag/core.py b/dask/bag/core.py
index c1bb53d9b..756b876c8 100644
--- a/dask/bag/core.py
+++ b/dask/bag/core.py
@@ -1601,7 +1601,10 @@ class Bag(DaskMethodsMixin):
         return [Delayed(k, dsk) for k in keys]
 
     def repartition(self, npartitions):
-        """ Coalesce bag into fewer partitions.
+        """ Changes the number of partitions of the bag.
+
+        This can be used to reduce or increase the number of partitions
+        of the bag.
 
         Examples
         --------
diff --git a/dask/dataframe/_compat.py b/dask/dataframe/_compat.py
new file mode 100644
index 000000000..16453b0ad
--- /dev/null
+++ b/dask/dataframe/_compat.py
@@ -0,0 +1,85 @@
+import string
+from distutils.version import LooseVersion
+
+import numpy as np
+import pandas as pd
+
+
+PANDAS_VERSION = LooseVersion(pd.__version__)
+PANDAS_GT_0230 = PANDAS_VERSION >= LooseVersion("0.23.0")
+PANDAS_GT_0240 = PANDAS_VERSION >= LooseVersion("0.24.0")
+PANDAS_GT_0250 = PANDAS_VERSION >= LooseVersion("0.25.0")
+PANDAS_GT_100 = PANDAS_VERSION >= LooseVersion("1.0.0rc0")
+HAS_INT_NA = PANDAS_GT_0240
+
+
+if PANDAS_GT_100:
+    import pandas.testing as tm  # noqa: F401
+else:
+    import pandas.util.testing as tm  # noqa: F401
+
+
+def assert_categorical_equal(left, right, *args, **kwargs):
+    if PANDAS_GT_100:
+        tm.assert_extension_array_equal(left, right, *args, **kwargs)
+        assert pd.api.types.is_categorical_dtype(
+            left.dtype
+        ), "{} is not categorical dtype".format(left)
+        assert pd.api.types.is_categorical_dtype(
+            right.dtype
+        ), "{} is not categorical dtype".format(right)
+    else:
+        return tm.assert_categorical_equal(left, right, *args, **kwargs)
+
+
+def assert_numpy_array_equal(left, right):
+    left_na = pd.isna(left)
+    right_na = pd.isna(right)
+    np.testing.assert_array_equal(left_na, right_na)
+
+    left_valid = left[~left_na]
+    right_valid = right[~right_na]
+    np.testing.assert_array_equal(left_valid, right_valid)
+
+
+def makeDataFrame():
+    data = np.random.randn(30, 4)
+    index = list(string.ascii_letters)[:30]
+    return pd.DataFrame(data, index=index, columns=list("ABCD"))
+
+
+def makeTimeDataFrame():
+    data = makeDataFrame()
+    data.index = makeDateIndex()
+    return data
+
+
+def makeTimeSeries():
+    return makeTimeDataFrame()["A"]
+
+
+def makeDateIndex(k=30, freq="B"):
+    return pd.date_range("2000", periods=k, freq=freq)
+
+
+def makeTimedeltaIndex(k=30, freq="D"):
+    return pd.timedelta_range("1 day", periods=k, freq=freq)
+
+
+def makeMissingDataframe():
+    df = makeDataFrame()
+    data = df.values
+    data = np.where(data > 1, np.nan, data)
+    return pd.DataFrame(data, index=df.index, columns=df.columns)
+
+
+def makeMixedDataFrame():
+    df = pd.DataFrame(
+        {
+            "A": [0.0, 1, 2, 3, 4],
+            "B": [0.0, 1, 0, 1, 0],
+            "C": ["foo{}".format(i) for i in range(5)],
+            "D": pd.date_range("2009-01-01", periods=5),
+        }
+    )
+    return df
diff --git a/dask/dataframe/core.py b/dask/dataframe/core.py
index 4052e9972..189e20632 100644
--- a/dask/dataframe/core.py
+++ b/dask/dataframe/core.py
@@ -413,9 +413,13 @@ class _Frame(DaskMethodsMixin, OperatorMethodMixin):
 
     def __repr__(self):
         data = self._repr_data().to_string(max_rows=5, show_dimensions=False)
-        return """Dask {klass} Structure:
+        _str_fmt = """Dask {klass} Structure:
 {data}
-Dask Name: {name}, {task} tasks""".format(
+Dask Name: {name}, {task} tasks"""
+        if len(self.columns) == 0:
+            data = data.partition("\n")[-1].replace("Index", "Divisions")
+            _str_fmt = "Empty {}".format(_str_fmt)
+        return _str_fmt.format(
             klass=self.__class__.__name__,
             data=data,
             name=key_split(self._name),
@@ -1709,9 +1713,11 @@ Dask Name: {name}, {task} tasks""".format(
                 self._meta_nonempty.select_dtypes(include=[np.timedelta64]).columns
             )
 
-            if count_timedeltas == len(self._meta.columns):
+            # pandas 1.0+ does not implement var on timedelta
+
+            if not PANDAS_GT_100 and count_timedeltas == len(self._meta.columns):
                 result = self._var_timedeltas(skipna, ddof, split_every)
-            elif count_timedeltas > 0:
+            elif not PANDAS_GT_100 and count_timedeltas > 0:
                 result = self._var_mixed(skipna, ddof, split_every)
             else:
                 result = self._var_numeric(skipna, ddof, split_every)
@@ -2109,7 +2115,11 @@ Dask Name: {name}, {task} tasks""".format(
 
             name2 = "{0}{1}-take-last".format(self._token_prefix, op_name)
             cumlast = map_partitions(
-                _take_last, cumpart, skipna, meta=pd.Series([]), token=name2
+                _take_last,
+                cumpart,
+                skipna,
+                meta=pd.Series([], dtype="float"),
+                token=name2,
             )
 
             suffix = tokenize(self)
@@ -2127,6 +2137,7 @@ Dask Name: {name}, {task} tasks""".format(
                 else:
                     # aggregate with previous cumulation results
                     layer[(cname, i)] = (
+                        methods._cum_aggregate_apply,
                         aggregate,
                         (cname, i - 1),
                         (cumlast._name, i - 1),
@@ -2336,7 +2347,12 @@ Dask Name: {name}, {task} tasks""".format(
         date = self.divisions[0] + offset
         end = self.loc._get_partitions(date)
 
-        include_right = offset.isAnchored() or not hasattr(offset, "_inc")
+        if PANDAS_GT_100:
+            is_anchored = offset.is_anchored()
+        else:
+            is_anchored = offset.isAnchored()
+
+        include_right = is_anchored or not hasattr(offset, "_inc")
 
         if end == self.npartitions - 1:
             divs = self.divisions
@@ -4285,8 +4301,14 @@ class DataFrame(_Frame):
     def _repr_data(self):
         meta = self._meta
         index = self._repr_divisions
-        series_list = [_repr_data_series(s, index=index) for _, s in meta.iteritems()]
-        return pd.concat(series_list, axis=1)
+        cols = meta.columns
+        if len(cols) == 0:
+            series_df = pd.DataFrame([[]] * len(index), columns=cols, index=index)
+        else:
+            series_df = pd.concat(
+                [_repr_data_series(s, index=index) for _, s in meta.iteritems()], axis=1
+            )
+        return series_df
 
     _HTML_FMT = """<div><strong>Dask DataFrame Structure:</strong></div>
 {data}
@@ -5113,7 +5135,7 @@ def quantile(df, q, method="default"):
         name = "quantiles-" + token
         empty_index = pd.Index([], dtype=float)
         return Series(
-            {(name, 0): pd.Series([], name=df.name, index=empty_index)},
+            {(name, 0): pd.Series([], name=df.name, index=empty_index, dtype="float")},
             name,
             df._meta,
             [None, None],
@@ -5386,7 +5408,7 @@ def _take_last(a, skipna=True):
             # create Series from appropriate backend dataframe library
             series_typ = type(a.iloc[0:1, 0])
             if a.empty:
-                return series_typ([])
+                return series_typ([], dtype="float")
             return series_typ(
                 {col: _last_valid(a[col]) for col in a.columns}, index=a.columns
             )
@@ -5886,13 +5908,18 @@ def maybe_shift_divisions(df, periods, freq):
     """
     if isinstance(freq, str):
         freq = pd.tseries.frequencies.to_offset(freq)
-    if isinstance(freq, pd.DateOffset) and (
-        freq.isAnchored() or not hasattr(freq, "delta")
-    ):
-        # Can't infer divisions on relative or anchored offsets, as
-        # divisions may now split identical index value.
-        # (e.g. index_partitions = [[1, 2, 3], [3, 4, 5]])
-        return df.clear_divisions()
+
+    is_offset = isinstance(freq, pd.DateOffset)
+    if is_offset:
+        if PANDAS_GT_100:
+            is_anchored = freq.is_anchored()
+        else:
+            is_anchored = freq.isAnchored()
+        if is_anchored or not hasattr(freq, "delta"):
+            # Can't infer divisions on relative or anchored offsets, as
+            # divisions may now split identical index value.
+            # (e.g. index_partitions = [[1, 2, 3], [3, 4, 5]])
+            return df.clear_divisions()
     if df.known_divisions:
         divs = pd.Series(range(len(df.divisions)), index=df.divisions)
         divisions = divs.shift(periods, freq=freq).index
diff --git a/dask/dataframe/groupby.py b/dask/dataframe/groupby.py
index eae981e56..3c9165a3c 100644
--- a/dask/dataframe/groupby.py
+++ b/dask/dataframe/groupby.py
@@ -51,7 +51,7 @@ from ..highlevelgraph import HighLevelGraph
 # operates on matching partitions of frame-like objects passed as varargs.
 #
 # After the initial chunk step, the passed index is implicitly passed along to
-# subsequent operations as the index of the parittions. Groupby operations on
+# subsequent operations as the index of the partitions. Groupby operations on
 # the individual partitions can then access the index via the ``levels``
 # parameter of the ``groupby`` function. The correct argument is determined by
 # the ``_determine_levels`` function.
@@ -126,7 +126,7 @@ def _groupby_raise_unaligned(df, **kwargs):
     """Groupby, but raise if df and `by` key are unaligned.
 
     Pandas supports grouping by a column that doesn't align with the input
-    frame/series/index. However, the reindexing this causes doesn't seem to be
+    frame/series/index. However, the reindexing does not seem to be
     threadsafe, and can result in incorrect results. Since grouping by an
     unaligned key is generally a bad idea, we just error loudly in dask.
 
diff --git a/dask/dataframe/io/parquet/arrow.py b/dask/dataframe/io/parquet/arrow.py
index 708a3ab7b..dd39282b0 100644
--- a/dask/dataframe/io/parquet/arrow.py
+++ b/dask/dataframe/io/parquet/arrow.py
@@ -326,7 +326,12 @@ class ArrowEngine(Engine):
         fs, piece, columns, index, categories=(), partitions=(), **kwargs
     ):
         if isinstance(index, list):
-            columns += index
+            for level in index:
+                # unclear if we can use set ops here. I think the order matters.
+                # Need the membership test to avoid duplicating index when
+                # we slice with `columns` later on.
+                if level not in columns:
+                    columns.append(level)
         if isinstance(piece, str):
             # `piece` is a file-path string
             piece = pq.ParquetDatasetPiece(
@@ -342,6 +347,16 @@ class ArrowEngine(Engine):
                 open_file_func=partial(fs.open, mode="rb"),
             )
 
+        # Ensure `columns` and `partitions` do not overlap
+        columns_and_parts = columns.copy()
+        if columns_and_parts and partitions:
+            for part_name in partitions.partition_names:
+                if part_name in columns:
+                    columns.remove(part_name)
+                else:
+                    columns_and_parts.append(part_name)
+            columns = columns or None
+
         df = piece.read(
             columns=columns,
             partitions=partitions,
@@ -349,7 +364,7 @@ class ArrowEngine(Engine):
             use_threads=False,
             **kwargs.get("read", {}),
         ).to_pandas(categories=categories, use_threads=False, ignore_metadata=True)[
-            list(columns)
+            list(columns_and_parts)
         ]
 
         if index:
diff --git a/dask/dataframe/methods.py b/dask/dataframe/methods.py
index 1c263e513..ebd27902e 100644
--- a/dask/dataframe/methods.py
+++ b/dask/dataframe/methods.py
@@ -231,6 +231,23 @@ def describe_nonnumeric_aggregate(stats, name):
     return pd.Series(values, index=index, name=name)
 
 
+def _cum_aggregate_apply(aggregate, x, y):
+    """ Apply aggregation function within a cumulative aggregation
+
+    Parameters
+    ----------
+    aggregate: function (a, a) -> a
+        The aggregation function, like add, which is used to and subsequent
+        results
+    x:
+    y:
+    """
+    if y is None:
+        return x
+    else:
+        return aggregate(x, y)
+
+
 def cummin_aggregate(x, y):
     if is_series_like(x) or is_dataframe_like(x):
         return x.where((x < y) | x.isnull(), y, axis=x.ndim - 1)
@@ -365,6 +382,9 @@ def concat_pandas(dfs, axis=0, join="outer", uniform=False, filter_warning=True)
     # Support concatenating indices along axis 0
     if isinstance(dfs[0], pd.Index):
         if isinstance(dfs[0], pd.CategoricalIndex):
+            for i in range(1, len(dfs)):
+                if not isinstance(dfs[i], pd.CategoricalIndex):
+                    dfs[i] = dfs[i].astype("category")
             return pd.CategoricalIndex(union_categoricals(dfs), name=dfs[0].name)
         elif isinstance(dfs[0], pd.MultiIndex):
             first, rest = dfs[0], dfs[1:]
diff --git a/dask/dataframe/multi.py b/dask/dataframe/multi.py
index 598176e09..1f15a1c7d 100644
--- a/dask/dataframe/multi.py
+++ b/dask/dataframe/multi.py
@@ -64,6 +64,7 @@ from pandas.api.types import is_dtype_equal
 from ..base import tokenize, is_dask_collection
 from ..highlevelgraph import HighLevelGraph
 from ..utils import apply
+from ._compat import PANDAS_GT_100
 from .core import (
     _Frame,
     DataFrame,
@@ -715,7 +716,10 @@ def concat_and_unsort(frames, columns):
 
 
 def _concat_compat(frames, left, right):
-    if PANDAS_GT_0230:
+    if PANDAS_GT_100:
+        # join_axes removed
+        return (pd.concat, frames, 0, "outer", False, None, None, None, False, False)
+    elif PANDAS_GT_0230:
         # (axis, join, join_axis, ignore_index, keys, levels, names, verify_integrity, sort)
         # we only care about sort, to silence warnings.
         return (
diff --git a/dask/dataframe/utils.py b/dask/dataframe/utils.py
index b77f14582..93b0c8de8 100644
--- a/dask/dataframe/utils.py
+++ b/dask/dataframe/utils.py
@@ -3,7 +3,6 @@ import numbers
 import re
 import textwrap
 from collections.abc import Iterator, Mapping
-from distutils.version import LooseVersion
 
 import sys
 import traceback
@@ -11,7 +10,6 @@ from contextlib import contextmanager
 
 import numpy as np
 import pandas as pd
-import pandas.util.testing as tm
 from pandas.api.types import (
     is_categorical_dtype,
     is_scalar,
@@ -21,6 +19,17 @@ from pandas.api.types import (
     is_interval_dtype,
 )
 
+# include these here for compat
+from ._compat import (  # noqa: F401
+    PANDAS_VERSION,
+    PANDAS_GT_0230,
+    PANDAS_GT_0240,
+    PANDAS_GT_0250,
+    PANDAS_GT_100,
+    HAS_INT_NA,
+    tm,
+)
+
 from .extensions import make_array_nonempty, make_scalar
 from ..base import is_dask_collection
 from ..core import get_deps
@@ -31,14 +40,6 @@ from ..utils import is_series_like as dask_is_series_like
 from ..utils import is_index_like as dask_is_index_like
 
 
-PANDAS_VERSION = LooseVersion(pd.__version__)
-PANDAS_GT_0230 = PANDAS_VERSION >= LooseVersion("0.23.0")
-PANDAS_GT_0240 = PANDAS_VERSION >= LooseVersion("0.24.0rc1")
-PANDAS_GT_0250 = PANDAS_VERSION >= LooseVersion("0.25.0")
-PANDAS_GT_100 = PANDAS_VERSION >= LooseVersion("0.26.0.dev0")
-HAS_INT_NA = PANDAS_GT_0240
-
-
 def is_integer_na_dtype(t):
     dtype = getattr(t, "dtype", t)
     if HAS_INT_NA:
diff --git a/dask/order.py b/dask/order.py
index 15a8fe52b..8aad68d7b 100644
--- a/dask/order.py
+++ b/dask/order.py
@@ -74,7 +74,8 @@ Work towards *small goals* with *big steps*.
     This relies on the regularity of graph constructors like dask.array to be a
     good proxy for ordering.  This is usually a good idea and a sane default.
 """
-from .core import get_dependencies, reverse_dict, get_deps  # noqa: F401
+from math import log
+from .core import get_dependencies, reverse_dict, get_deps, getcycle  # noqa: F401
 from .utils_test import add, inc  # noqa: F401
 
 
@@ -88,12 +89,12 @@ def order(dsk, dependencies=None):
     This currently traverses the graph as a single-threaded scheduler would
     traverse it.  It breaks ties in the following ways:
 
-    1.  Start from roots nodes that have the largest subgraphs
-    2.  When a node has dependencies that are not yet computed prefer
-        dependencies with large subtrees  (start hard things first)
-    2.  When we reach a node that can be computed we then traverse up and
-        prefer dependents that have small super-trees (few total dependents)
-        (finish existing work quickly)
+    1.  Begin at a leaf node that is a dependency of a root node that has the
+        largest subgraph (start hard things first)
+    2.  Prefer tall branches with few dependents (start hard things first and
+        try to avoid memory usage)
+    3.  Prefer dependents that are dependencies of root nodes that have
+        the smallest subgraph (do small goals that can terminate quickly)
 
     Examples
     --------
@@ -101,115 +102,381 @@ def order(dsk, dependencies=None):
     >>> order(dsk)
     {'a': 0, 'c': 1, 'b': 2, 'd': 3}
     """
+    if not dsk:
+        return {}
+
     if dependencies is None:
         dependencies = {k: get_dependencies(dsk, k) for k in dsk}
 
-    for k, deps in dependencies.items():
-        deps.discard(k)
-
     dependents = reverse_dict(dependencies)
+    num_needed, total_dependencies = ndependencies(dependencies, dependents)
+    metrics = graph_metrics(dependencies, dependents, total_dependencies)
+    if len(metrics) != len(dsk):
+        cycle = getcycle(dsk, None)
+        raise RuntimeError(
+            "Cycle detected between the following keys:\n  -> %s"
+            % "\n  -> ".join(str(x) for x in cycle)
+        )
+
+    # Leaf nodes.  We choose one--the initial node--for each weakly connected subgraph.
+    # Let's calculate the `initial_stack_key` as we determine `init_stack` set.
+    init_stack = {
+        # First prioritize large, tall groups, then prioritize the same as ``dependents_key``.
+        key: (
+            # at a high-level, work towards a large goal (and prefer tall and narrow)
+            -max_dependencies,
+            num_dependents - max_heights,
+            # tactically, finish small connected jobs first
+            min_dependencies,
+            num_dependents - min_heights,  # prefer tall and narrow
+            -total_dependents,  # take a big step
+            # try to be memory efficient
+            num_dependents,
+            # tie-breaker
+            StrComparable(key),
+        )
+        for key, num_dependents, (
+            total_dependents,
+            min_dependencies,
+            max_dependencies,
+            min_heights,
+            max_heights,
+        ) in (
+            (key, len(dependents[key]), metrics[key])
+            for key, val in dependencies.items()
+            if not val
+        )
+    }
+    # `initial_stack_key` chooses which task to run at the very beginning.
+    # This value is static, so we pre-compute as the value of this dict.
+    initial_stack_key = init_stack.__getitem__
 
-    total_dependencies = ndependencies(dependencies, dependents)
-    total_dependents, min_dependencies = ndependents(
-        dependencies, dependents, total_dependencies
-    )
-
-    waiting = {k: set(v) for k, v in dependencies.items()}
+    def dependents_key(x):
+        """ Choose a path from our starting task to our tactical goal
+
+        This path is connected to a large goal, but focuses on completing a small goal.
+        """
+        num_dependents = len(dependents[x])
+        total_dependents, min_dependencies, _, min_heights, _ = metrics[x]
+        return (
+            # tactically, finish small connected jobs first
+            min_dependencies,
+            num_dependents - min_heights,  # prefer tall and narrow
+            -total_dependents,  # take a big step
+            # try to be memory efficient
+            num_dependents - len(dependencies[x]) + num_needed[x],
+            num_dependents,
+            total_dependencies[x],  # already found work, so don't add more
+            # tie-breaker
+            StrComparable(x),
+        )
 
     def dependencies_key(x):
-        return total_dependencies.get(x, 0), ReverseStrComparable(x)
+        """ Choose which dependency to run as part of a reverse DFS
+
+        This is very similar to both ``initial_stack_key`` and ``dependents_key``.
+        """
+        num_dependents = len(dependents[x])
+        (
+            total_dependents,
+            min_dependencies,
+            max_dependencies,
+            min_heights,
+            max_heights,
+        ) = metrics[x]
+        # Prefer short and narrow instead of tall in narrow, because we're going in
+        # reverse along dependencies.
+        return (
+            # at a high-level, work towards a large goal (and prefer short and narrow)
+            -max_dependencies,
+            num_dependents + max_heights,
+            # tactically, finish small connected jobs first
+            min_dependencies,
+            num_dependents + min_heights,  # prefer short and narrow
+            -total_dependencies[x],  # go where the work is
+            # try to be memory efficient
+            num_dependents - len(dependencies[x]) + num_needed[x],
+            num_dependents,
+            total_dependents,  # already found work, so don't add more
+            # tie-breaker
+            StrComparable(x),
+        )
+
+    result = {}
+    i = 0
 
-    def dependents_key(x):
-        return (min_dependencies[x], -(total_dependents.get(x, 0)), StrComparable(x))
+    # Nodes in the current weakly connected subgraph that are not part of the current
+    # DFS along dependencies.  The next DFS will begin from these nodes.  `outer_stack`
+    # is populated from `init_stack` and the dependents of `outer_stack_seeds`.
+    outer_stack = []
+
+    # Nodes get added to this as they complete.  We will populate `outer_stack` from
+    # dependents of these nodes.  We consider nodes using a FIFO policy (yes, we could
+    # have used collections.deque, but a list with the current index is faster).
+    # Note that this does not include nodes with no dependents!
+    outer_stack_seeds = []
+    outer_stack_seeds_index = 0
+
+    # Used to perform a DFS along dependencies.  Once emptied (when traversing dependencies),
+    # this continues (along dependents) along a path from the initial (leaf) node down to a
+    # root node (a tactical goal).  This ensures this root node will get calculated before
+    # we consider any other dependents from `outer_stack_seeds`.
+    inner_stack = []
+
+    # aliases for speed
+    outer_stack_append = outer_stack.append
+    outer_stack_pop = outer_stack.pop
+    outer_stack_extend = outer_stack.extend
+    outer_stack_seeds_append = outer_stack_seeds.append
+    inner_stack_append = inner_stack.append
+    inner_stack_pop = inner_stack.pop
+    inner_stack_extend = inner_stack.extend
+    set_difference = set.difference
+
+    is_init_sorted = False
+    item = min(init_stack, key=initial_stack_key)
+    while True:
+        outer_stack_append(item)
+        while outer_stack:
+            item = outer_stack_pop()
+            if item not in result:
+                inner_stack_append(item)
+
+                while inner_stack:
+                    # Perform a DFS along dependencies until we complete our tactical goal
+                    item = inner_stack_pop()
+                    if item in result:
+                        continue
+
+                    if num_needed[item]:
+                        inner_stack_append(item)
+                        deps = set_difference(dependencies[item], result)
+                        if 1 < len(deps) < 1000:
+                            inner_stack_extend(
+                                sorted(deps, key=dependencies_key, reverse=True)
+                            )
+                        else:
+                            inner_stack_extend(deps)
+                        continue
+
+                    result[item] = i
+                    i += 1
+                    deps = dependents[item]
+
+                    if metrics[item][3] == 2:  # min_height
+                        # Don't leave any dangling single nodes!  Finish all dependents that are
+                        # ready and are also root nodes.  Doing this here also lets us continue
+                        # down to a different root node.
+                        finish_now = {
+                            dep
+                            for dep in deps
+                            if not dependents[dep] and num_needed[dep] == 1
+                        }
+                        if finish_now:
+                            deps -= finish_now
+                            if len(finish_now) > 1:
+                                finish_now = sorted(finish_now, key=dependents_key)
+                            for dep in finish_now:
+                                result[dep] = i
+                                i += 1
+
+                    if deps:
+                        for dep in deps:
+                            num_needed[dep] -= 1
+                        if not inner_stack:
+                            # Continue towards our tactical goal (an easy-to-compute root)
+                            if len(deps) == 1:
+                                inner_stack_extend(deps)
+                            else:
+                                # If there are many, many dependents, then calculating
+                                # `dependents_key` here could be relatively expensive.
+                                # However, this is still probably worth it tactically.
+                                inner_stack_append(min(deps, key=dependents_key))
+                                outer_stack_seeds_append(item)
+                        else:
+                            # Our next starting point will be "seeded" by a completed task in a
+                            # FIFO manner.  When our DFS with `inner_stack` is finished--which
+                            # means we computed our tactical goal--we will choose our next starting
+                            # point from the dependents of completed tasks.  However, it is too
+                            # expensive to consider all dependents of all completed tasks, so we
+                            # consider the dependents of tasks in the order they complete.
+                            outer_stack_seeds_append(item)
+
+            while not outer_stack and outer_stack_seeds_index < len(outer_stack_seeds):
+                # We wait for as long as possible to consider dependents of completed nodes:
+                # 1. some may have already completed, so waiting lets us sort fewer items, and
+                # 2. sorting via `dependents_key` depends on `num_needed`, which is dynamic.
+                deps = set_difference(
+                    dependents[outer_stack_seeds[outer_stack_seeds_index]], result
+                )
+                if deps:
+                    if 1 < len(deps) < 1000:
+                        outer_stack_extend(
+                            sorted(deps, key=dependents_key, reverse=True)
+                        )
+                    else:
+                        outer_stack_extend(deps)
+                outer_stack_seeds_index += 1
+
+        if len(dependencies) == len(result):
+            break  # all done!
+
+        # We just finished computing a connected group.
+        # Let's choose the first `item` in the next group to compute.
+        # If we have few large groups left, then it's best to find `item` by taking a minimum.
+        # If we have many small groups left, then it's best to sort.
+        # If we have many tiny groups left, then it's best to simply iterate.
+        if not is_init_sorted:
+            prev_len = len(init_stack)
+            if type(init_stack) is dict:
+                init_stack = set(init_stack)
+            init_stack = set_difference(init_stack, result)
+            N = len(init_stack)
+            m = prev_len - N
+            # is `min` likely better than `sort`?
+            if m >= N or N + (N - m) * log(N - m) < N * log(N):
+                item = min(init_stack, key=initial_stack_key)
+                continue
+
+            if len(init_stack) < 10000:
+                init_stack = sorted(init_stack, key=initial_stack_key, reverse=True)
+            else:
+                init_stack = list(init_stack)
+            init_stack_pop = init_stack.pop
+            is_init_sorted = True
+
+        item = init_stack_pop()
+        while item in result:
+            item = init_stack_pop()
 
-    result = dict()
-    seen = set()  # tasks that should not be added again to the stack
-    i = 0
+    return result
 
-    stack = [k for k, v in dependents.items() if not v]
-    if len(stack) < 10000:
-        stack = sorted(stack, key=dependencies_key)
-    else:
-        stack = stack[::-1]
 
-    while stack:
-        item = stack.pop()
+def graph_metrics(dependencies, dependents, total_dependencies):
+    r""" Useful measures of a graph used by ``dask.order.order``
 
-        if item in result:
-            continue
+    Example DAG (a1 has no dependencies; b2 and c1 are root nodes):
 
-        deps = waiting[item]
+    c1
+    |
+    b1  b2
+     \  /
+      a1
 
-        if deps:
-            stack.append(item)
-            seen.add(item)
-            if len(deps) < 1000:
-                deps = sorted(deps, key=dependencies_key)
-            stack.extend(deps)
-            continue
+    For each key we return:
+    1.  The number of keys that can only be run after this key is run.  The
+        root nodes have value 1 while deep child nodes will have larger values.
 
-        result[item] = i
-        i += 1
+        1
+        |
+        2   1
+         \ /
+          4
 
-        for dep in dependents[item]:
-            waiting[dep].discard(item)
+    2.  The minimum value of the total number of dependencies of
+        all final dependents (see module-level comment for more).
+        In other words, the minimum of ``ndependencies`` of root
+        nodes connected to the current node.
 
-        deps = [
-            d
-            for d in dependents[item]
-            if d not in result and not (d in seen and len(waiting[d]) > 1)
-        ]
-        if len(deps) < 1000:
-            deps = sorted(deps, key=dependents_key, reverse=True)
+        3
+        |
+        3   2
+         \ /
+          2
 
-        stack.extend(deps)
+    3.  The maximum value of the total number of dependencies of
+        all final dependents (see module-level comment for more).
+        In other words, the maximum of ``ndependencies`` of root
+        nodes connected to the current node.
 
-    return result
+        3
+        |
+        3   2
+         \ /
+          3
 
+    4.  The minimum height from a root node
 
-def ndependents(dependencies, dependents, total_dependencies):
-    """ Number of total data elements that depend on key
+        1
+        |
+        2   1
+         \ /
+          2
 
-    For each key we return the number of keys that can only be run after this
-    key is run.  The root nodes have value 1 while deep child nodes will have
-    larger values.
+    5.  The maximum height from a root node
 
-    We also return the minimum value of the maximum number of dependencies of
-    all final dependencies (see module-level comment for more)
+        1
+        |
+        2   1
+         \ /
+          3
 
     Examples
     --------
-    >>> dsk = {'a': 1, 'b': (inc, 'a'), 'c': (inc, 'b')}
+    >>> dsk = {'a1': 1, 'b1': (inc, 'a1'), 'b2': (inc, 'a1'), 'c1': (inc, 'b1')}
     >>> dependencies, dependents = get_deps(dsk)
-
-    >>> total_dependencies = ndependencies(dependencies, dependents)
-    >>> total_dependents, min_dependencies = ndependents(dependencies,
-    ...                                                  dependents,
-    ...                                                  total_dependencies)
-    >>> sorted(total_dependents.items())
-    [('a', 3), ('b', 2), ('c', 1)]
+    >>> _, total_dependencies = ndependencies(dependencies, dependents)
+    >>> metrics = graph_metrics(dependencies, dependents, total_dependencies)
+    >>> sorted(metrics.items())
+    [('a1', (4, 2, 3, 2, 3)), ('b1', (2, 3, 3, 2, 2)), ('b2', (1, 2, 2, 1, 1)), ('c1', (1, 3, 3, 1, 1))]
 
     Returns
     -------
-    total_dependendents: Dict[key, int]
-    min_dependencies: Dict[key, int]
+    metrics: Dict[key, Tuple[int, int, int, int, int]]
     """
-    result = dict()
-    min_result = dict()
-    num_needed = {k: len(v) for k, v in dependents.items()}
-    current = {k for k, v in num_needed.items() if v == 0}
+    result = {}
+    num_needed = {k: len(v) for k, v in dependents.items() if v}
+    current = []
+    current_pop = current.pop
+    current_append = current.append
+    for key, deps in dependents.items():
+        if not deps:
+            val = total_dependencies[key]
+            result[key] = (1, val, val, 1, 1)
+            for child in dependencies[key]:
+                num_needed[child] -= 1
+                if not num_needed[child]:
+                    current_append(child)
+
     while current:
-        key = current.pop()
-        result[key] = 1 + sum(result[parent] for parent in dependents[key])
-        try:
-            min_result[key] = min(min_result[parent] for parent in dependents[key])
-        except ValueError:
-            min_result[key] = total_dependencies[key]
+        key = current_pop()
+        parents = dependents[key]
+        if len(parents) == 1:
+            (parent,) = parents
+            (
+                total_dependents,
+                min_dependencies,
+                max_dependencies,
+                min_heights,
+                max_heights,
+            ) = result[parent]
+            result[key] = (
+                1 + total_dependents,
+                min_dependencies,
+                max_dependencies,
+                1 + min_heights,
+                1 + max_heights,
+            )
+        else:
+            (
+                total_dependents,
+                min_dependencies,
+                max_dependencies,
+                min_heights,
+                max_heights,
+            ) = zip(*(result[parent] for parent in dependents[key]))
+            result[key] = (
+                1 + sum(total_dependents),
+                min(min_dependencies),
+                max(max_dependencies),
+                1 + min(min_heights),
+                1 + max(max_heights),
+            )
         for child in dependencies[key]:
             num_needed[child] -= 1
-            if num_needed[child] == 0:
-                current.add(child)
-    return result, min_result
+            if not num_needed[child]:
+                current_append(child)
+    return result
 
 
 def ndependencies(dependencies, dependents):
@@ -222,20 +489,34 @@ def ndependencies(dependencies, dependents):
     --------
     >>> dsk = {'a': 1, 'b': (inc, 'a'), 'c': (inc, 'b')}
     >>> dependencies, dependents = get_deps(dsk)
-    >>> sorted(ndependencies(dependencies, dependents).items())
+    >>> num_dependencies, total_dependencies = ndependencies(dependencies, dependents)
+    >>> sorted(total_dependencies.items())
     [('a', 1), ('b', 2), ('c', 3)]
+
+    Returns
+    -------
+    num_dependencies: Dict[key, int]
+    total_dependencies: Dict[key, int]
     """
-    result = dict()
     num_needed = {k: len(v) for k, v in dependencies.items()}
-    current = {k for k, v in num_needed.items() if v == 0}
+    num_dependencies = num_needed.copy()
+    current = []
+    current_pop = current.pop
+    current_append = current.append
+    result = {k: 1 for k, v in dependencies.items() if not v}
+    for key in result:
+        for parent in dependents[key]:
+            num_needed[parent] -= 1
+            if not num_needed[parent]:
+                current_append(parent)
     while current:
-        key = current.pop()
+        key = current_pop()
         result[key] = 1 + sum(result[child] for child in dependencies[key])
         for parent in dependents[key]:
             num_needed[parent] -= 1
-            if num_needed[parent] == 0:
-                current.add(parent)
-    return result
+            if not num_needed[parent]:
+                current_append(parent)
+    return num_dependencies, result
 
 
 class StrComparable(object):
@@ -265,22 +546,3 @@ class StrComparable(object):
             return self.obj < other.obj
         except Exception:
             return str(self.obj) < str(other.obj)
-
-
-class ReverseStrComparable(object):
-    """ Wrap object so that it defaults to string comparison
-
-    Used when sorting in reverse direction.  See StrComparable for normal
-    documentation.
-    """
-
-    __slots__ = ("obj",)
-
-    def __init__(self, obj):
-        self.obj = obj
-
-    def __lt__(self, other):
-        try:
-            return self.obj > other.obj
-        except Exception:
-            return str(self.obj) > str(other.obj)
diff --git a/docs/source/custom-collections.rst b/docs/source/custom-collections.rst
index ba31ea684..30e5459b7 100644
--- a/docs/source/custom-collections.rst
+++ b/docs/source/custom-collections.rst
@@ -560,9 +560,12 @@ implementation. There are two ways to do this:
 
 2. Register a function with ``dask.base.normalize_token``
 
-   If defining a method on the class isn't possible, you can register a tokenize
-   function with the ``normalize_token`` dispatch.  The function should have the
-   same signature as described above.
+   If defining a method on the class isn't possible or you need to
+   customize the tokenize function for a class whose super-class is
+   already registered (for example if you need to sub-class built-ins),
+   you can register a tokenize function with the ``normalize_token``
+   dispatch.  The function should have the same signature as described
+   above.
 
 In both cases the implementation should be the same, where only the location of the
 definition is different.
diff --git a/docs/source/futures.rst b/docs/source/futures.rst
index fcc1f08c1..5b63f69ea 100644
--- a/docs/source/futures.rst
+++ b/docs/source/futures.rst
@@ -108,6 +108,9 @@ However, note that each task comes with about 1ms of overhead.  If you want to
 map a function over a large number of inputs, then you might consider
 :doc:`dask.bag <bag>` or :doc:`dask.dataframe <dataframe>` instead.
 
+.. note: See `this page <https://docs.dask.org/en/latest/graphs.html>`_ for
+   restrictions on what functions you use with Dask.
+
 Move Data
 ---------
 
diff --git a/docs/source/graphs.rst b/docs/source/graphs.rst
index f54f818fd..620e0ec0e 100644
--- a/docs/source/graphs.rst
+++ b/docs/source/graphs.rst
@@ -97,3 +97,51 @@ special and others can write different schedulers better suited to other
 applications or architectures easily.  Systems that emit dask graphs (like
 Dask Array, Dask Bag, and so on) may leverage the appropriate scheduler for
 the application and hardware.
+
+
+Task Expectations
+-----------------
+
+When a task is submitted to Dask for execution, there are a number of assumptions
+that are made about that task.
+
+Don't Modify Data In-Place
+~~~~~~~~~~~~~~~~~~~~~~~~~~
+
+In general, tasks with side-effects that alter the state of a future in-place
+are not recommended. Modifying data that is stored in Dask in-place can have
+unintended consequences. For example, consider a workflow involving a Numpy
+array:
+
+.. code-block:: python
+
+   from dask.distributed import Client
+   import numpy as np
+
+   client = Client()
+   x = client.submit(np.arange, 10)  # [0, 1, 2, 3, ...]
+
+   def f(arr):
+       arr[arr > 5] = 0  # modifies input directly without making a copy
+       arr += 1          # modifies input directly without making a copy
+       return arr
+
+   y = client.submit(f, x)
+
+In the example above Dask will update the values of the Numpy array
+``x`` in-place.  While efficient, this behavior can have unintended consequences,
+particularly if other tasks need to use ``x``, or if Dask needs to rerun this
+computation multiple times because of worker failure.
+
+
+Avoid Holding the GIL
+~~~~~~~~~~~~~~~~~~~~~
+
+Some Python functions that wrap external C/C++ code can hold onto the GIL,
+which stops other Python code from running in the background.  This is
+troublesome because while Dask workers run your function, they also need to
+communicate to each other in the background.
+
+If you wrap external code then please try to release the GIL.  This is usually
+easy to do if you are using any of the common solutions to code-wrapping like
+Cython, Numba, ctypes or others.


Test patch:
diff --git a/dask/array/tests/test_array_core.py b/dask/array/tests/test_array_core.py
index bfe9b1e1f..f5a774aaa 100644
--- a/dask/array/tests/test_array_core.py
+++ b/dask/array/tests/test_array_core.py
@@ -971,6 +971,20 @@ def test_broadcast_arrays():
         assert_eq(e_a_r, e_d_r)
 
 
+def test_broadcast_arrays_uneven_chunks():
+    x = da.ones(30, chunks=(3,))
+    y = da.ones(30, chunks=(5,))
+    z = np.broadcast_arrays(x, y)
+
+    assert_eq(z, z)
+
+    x = da.ones((1, 30), chunks=(1, 3))
+    y = da.ones(30, chunks=(5,))
+    z = np.broadcast_arrays(x, y)
+
+    assert_eq(z, z)
+
+
 @pytest.mark.parametrize(
     "u_shape, v_shape",
     [
diff --git a/dask/bytes/tests/test_s3.py b/dask/bytes/tests/test_s3.py
index 871e54de9..050ba5ccb 100644
--- a/dask/bytes/tests/test_s3.py
+++ b/dask/bytes/tests/test_s3.py
@@ -386,6 +386,8 @@ def test_modification_time_read_bytes():
 @pytest.mark.parametrize("engine", ["pyarrow", "fastparquet"])
 def test_parquet(s3, engine):
     dd = pytest.importorskip("dask.dataframe")
+    from dask.dataframe._compat import tm
+
     lib = pytest.importorskip(engine)
     if engine == "pyarrow" and LooseVersion(lib.__version__) < "0.13.1":
         pytest.skip("pyarrow < 0.13.1 not supported for parquet")
@@ -415,7 +417,7 @@ def test_parquet(s3, engine):
     df2 = dd.read_parquet(url, index="foo", engine=engine)
     assert len(df2.divisions) > 1
 
-    pd.util.testing.assert_frame_equal(data, df2.compute())
+    tm.assert_frame_equal(data, df2.compute())
 
 
 def test_parquet_wstoragepars(s3):
diff --git a/dask/dataframe/io/tests/test_csv.py b/dask/dataframe/io/tests/test_csv.py
index 50ec4b7ab..fbf0bec12 100644
--- a/dask/dataframe/io/tests/test_csv.py
+++ b/dask/dataframe/io/tests/test_csv.py
@@ -11,10 +11,9 @@ dd = pytest.importorskip("dask.dataframe")
 
 from toolz import partition_all, valmap
 
-import pandas.util.testing as tm
-
 import dask
 import dask.dataframe as dd
+from dask.dataframe._compat import tm
 from dask.base import compute_as_if_collection
 from dask.dataframe.io.csv import (
     text_blocks_to_pandas,
@@ -673,7 +672,7 @@ def test_compression_multiple_files():
         f.write(csv_text.encode())
         f.close()
 
-        with tm.assert_produces_warning(UserWarning):
+        with pytest.warns(UserWarning):
             df = dd.read_csv(os.path.join(tdir, "*.csv.gz"), compression="gzip")
 
         assert len(df.compute()) == (len(csv_text.split("\n")) - 1) * 2
diff --git a/dask/dataframe/io/tests/test_demo.py b/dask/dataframe/io/tests/test_demo.py
index c6d0d7e00..f8bbb11a7 100644
--- a/dask/dataframe/io/tests/test_demo.py
+++ b/dask/dataframe/io/tests/test_demo.py
@@ -1,9 +1,9 @@
-import pandas.util.testing as tm
 import pandas as pd
 import pytest
 
 import dask
 import dask.dataframe as dd
+from dask.dataframe._compat import tm
 from dask.dataframe.utils import assert_eq
 
 
diff --git a/dask/dataframe/io/tests/test_hdf.py b/dask/dataframe/io/tests/test_hdf.py
index 9909c5990..0325860fb 100644
--- a/dask/dataframe/io/tests/test_hdf.py
+++ b/dask/dataframe/io/tests/test_hdf.py
@@ -1,6 +1,5 @@
 import numpy as np
 import pandas as pd
-import pandas.util.testing as tm
 import pytest
 
 import os
@@ -9,10 +8,16 @@ import pathlib
 
 import dask
 import dask.dataframe as dd
+from dask.dataframe._compat import tm
 from dask.utils import tmpfile, tmpdir, dependency_depth
 from dask.dataframe.utils import assert_eq
 
 
+skip_pandas_rc = pytest.mark.xfail(
+    dd._compat.PANDAS_GT_100, reason="https://github.com/pandas-dev/pandas/issues/30962"
+)
+
+
 def test_to_hdf():
     pytest.importorskip("tables")
     df = pd.DataFrame(
@@ -44,6 +49,7 @@ def test_to_hdf():
         tm.assert_frame_equal(df, out[:])
 
 
+@skip_pandas_rc
 def test_to_hdf_multiple_nodes():
     pytest.importorskip("tables")
     df = pd.DataFrame(
@@ -124,6 +130,7 @@ def test_to_hdf_multiple_nodes():
             assert_eq(df16, out)
 
 
+@skip_pandas_rc
 def test_to_hdf_multiple_files():
     pytest.importorskip("tables")
     df = pd.DataFrame(
@@ -207,6 +214,7 @@ def test_to_hdf_multiple_files():
             assert_eq(df, out)
 
 
+@skip_pandas_rc
 def test_to_hdf_modes_multiple_nodes():
     pytest.importorskip("tables")
     df = pd.DataFrame(
@@ -255,6 +263,7 @@ def test_to_hdf_modes_multiple_nodes():
         assert_eq(df.append(df), out)
 
 
+@skip_pandas_rc
 def test_to_hdf_modes_multiple_files():
     pytest.importorskip("tables")
     df = pd.DataFrame(
@@ -367,6 +376,7 @@ def test_to_hdf_link_optimizations():
 
 
 @pytest.mark.slow
+@skip_pandas_rc
 def test_to_hdf_lock_delays():
     pytest.importorskip("tables")
     df16 = pd.DataFrame(
@@ -458,6 +468,7 @@ def test_to_hdf_exceptions():
 
 @pytest.mark.parametrize("scheduler", ["sync", "threads", "processes"])
 @pytest.mark.parametrize("npartitions", [1, 4, 10])
+@skip_pandas_rc
 def test_to_hdf_schedulers(scheduler, npartitions):
     pytest.importorskip("tables")
     df = pd.DataFrame(
@@ -609,6 +620,7 @@ def test_to_fmt_warns():
         (pd.Series([1, 2, 3, 4], name="a"), tm.assert_series_equal),
     ],
 )
+@skip_pandas_rc
 def test_read_hdf(data, compare):
     pytest.importorskip("tables")
     with tmpfile("h5") as fn:
@@ -644,6 +656,7 @@ def test_read_hdf(data, compare):
         compare(a.compute(), sorted_data)
 
 
+@skip_pandas_rc
 def test_read_hdf_multiply_open():
     """Test that we can read from a file that's already opened elsewhere in
     read-only mode."""
@@ -657,6 +670,7 @@ def test_read_hdf_multiply_open():
             dd.read_hdf(fn, "/data", chunksize=2, mode="r")
 
 
+@skip_pandas_rc
 def test_read_hdf_multiple():
     pytest.importorskip("tables")
     df = pd.DataFrame(
@@ -710,6 +724,7 @@ def test_read_hdf_multiple():
         assert_eq(a, r)
 
 
+@skip_pandas_rc
 def test_read_hdf_start_stop_values():
     pytest.importorskip("tables")
     df = pd.DataFrame(
@@ -728,6 +743,7 @@ def test_read_hdf_start_stop_values():
             dd.read_hdf(fn, "/data", chunksize=-1)
 
 
+@skip_pandas_rc
 def test_hdf_globbing():
     pytest.importorskip("tables")
     df = pd.DataFrame(
@@ -765,6 +781,7 @@ def test_hdf_globbing():
             tm.assert_frame_equal(res.compute(), pd.concat([df] * 3))
 
 
+@skip_pandas_rc
 def test_hdf_file_list():
     pytest.importorskip("tables")
     df = pd.DataFrame(
@@ -781,6 +798,7 @@ def test_hdf_file_list():
             tm.assert_frame_equal(res.compute(), df)
 
 
+@skip_pandas_rc
 def test_read_hdf_pattern_pathlike():
     pytest.importorskip("tables")
     df = pd.DataFrame(
@@ -808,6 +826,7 @@ def test_to_hdf_path_pathlike():
         assert_eq(res, ddf)
 
 
+@skip_pandas_rc
 def test_read_hdf_doesnt_segfault():
     pytest.importorskip("tables")
     with tmpfile("h5") as fn:
diff --git a/dask/dataframe/io/tests/test_io.py b/dask/dataframe/io/tests/test_io.py
index 6f10b2a6c..e591fd11b 100644
--- a/dask/dataframe/io/tests/test_io.py
+++ b/dask/dataframe/io/tests/test_io.py
@@ -1,6 +1,5 @@
 import numpy as np
 import pandas as pd
-import pandas.util.testing as tm
 
 import pytest
 from threading import Lock
@@ -8,6 +7,7 @@ from multiprocessing.pool import ThreadPool
 
 import dask.array as da
 import dask.dataframe as dd
+from dask.dataframe._compat import tm
 from dask.dataframe.io.io import _meta_from_array
 from dask.delayed import Delayed, delayed
 
@@ -263,7 +263,7 @@ def test_from_pandas_small():
             ddf = dd.from_pandas(df, npartitions=5, sort=sort)
             assert_eq(df, ddf)
 
-            s = pd.Series([0] * i, name="x")
+            s = pd.Series([0] * i, name="x", dtype=int)
             ds = dd.from_pandas(s, npartitions=5, sort=sort)
             assert_eq(s, ds)
 
diff --git a/dask/dataframe/io/tests/test_parquet.py b/dask/dataframe/io/tests/test_parquet.py
index 718216bfd..0a436b1c0 100644
--- a/dask/dataframe/io/tests/test_parquet.py
+++ b/dask/dataframe/io/tests/test_parquet.py
@@ -4,7 +4,6 @@ from distutils.version import LooseVersion
 
 import numpy as np
 import pandas as pd
-import pandas.util.testing as tm
 import pytest
 
 import dask
@@ -908,7 +907,7 @@ def test_empty_partition(tmpdir, engine):
 
 def test_timestamp_index(tmpdir, engine):
     fn = str(tmpdir)
-    df = tm.makeTimeDataFrame()
+    df = dd._compat.makeTimeDataFrame()
     df.index.name = "foo"
     ddf = dd.from_pandas(df, npartitions=5)
     ddf.to_parquet(fn, engine=engine)
@@ -2243,3 +2242,17 @@ def test_roundtrip_pandas_chunksize(tmpdir, write_engine, read_engine):
     )
 
     assert_eq(pdf, ddf_read)
+
+
+def test_read_pandas_fastparquet_partitioned(tmpdir, engine):
+    check_fastparquet()
+
+    pdf = pd.DataFrame(
+        [{"str": str(i), "int": i, "group": "ABC"[i % 3]} for i in range(6)]
+    )
+    path = str(tmpdir)
+    pdf.to_parquet(path, partition_cols=["group"], engine="fastparquet")
+    ddf_read = dd.read_parquet(path, engine=engine)
+
+    assert len(ddf_read["group"].compute()) == 6
+    assert len(ddf_read.compute().group) == 6
diff --git a/dask/dataframe/tests/test_accessors.py b/dask/dataframe/tests/test_accessors.py
index b86c7fbb4..aafae2b91 100644
--- a/dask/dataframe/tests/test_accessors.py
+++ b/dask/dataframe/tests/test_accessors.py
@@ -47,7 +47,7 @@ def test_register(obj, registrar):
     with ensure_removed(obj, "mine"):
         before = set(dir(obj))
         registrar("mine")(MyAccessor)
-        instance = dd.from_pandas(obj._partition_type([]), 2)
+        instance = dd.from_pandas(obj._partition_type([], dtype=float), 2)
         assert instance.mine.prop == "item"
         after = set(dir(obj))
         assert (before ^ after) == {"mine"}
diff --git a/dask/dataframe/tests/test_arithmetics_reduction.py b/dask/dataframe/tests/test_arithmetics_reduction.py
index bf4c8affa..6998afc03 100644
--- a/dask/dataframe/tests/test_arithmetics_reduction.py
+++ b/dask/dataframe/tests/test_arithmetics_reduction.py
@@ -7,6 +7,7 @@ import pandas as pd
 from pandas.api.types import is_datetime64_ns_dtype
 
 import dask.dataframe as dd
+from dask.dataframe._compat import PANDAS_GT_100
 from dask.dataframe.utils import (
     assert_eq,
     assert_dask_graph,
@@ -799,17 +800,6 @@ def test_reductions_timedelta(split_every):
     ds = pd.Series(pd.to_timedelta([2, 3, 4, np.nan, 5]))
     dds = dd.from_pandas(ds, 2)
 
-    with pytest.warns(None):
-        # runtime warnings; https://github.com/dask/dask/issues/2381
-        assert_eq(dds.var(split_every=split_every), ds.var())
-        if not PANDAS_GT_0250:
-            # https://github.com/pandas-dev/pandas/issues/18880
-            assert_eq(
-                dds.var(split_every=split_every, skipna=False), ds.var(skipna=False)
-            )
-
-    assert_eq(dds.var(ddof=0, split_every=split_every), ds.var(ddof=0))
-
     assert_eq(dds.sum(split_every=split_every), ds.sum())
     assert_eq(dds.min(split_every=split_every), ds.min())
     assert_eq(dds.max(split_every=split_every), ds.max())
@@ -822,12 +812,12 @@ def test_reductions_timedelta(split_every):
         (
             pd.DataFrame({"a": [1, 2, 3], "b": [4, 5, 6]}, index=[0, 1, 3]),
             0,
-            pd.Series([]),
+            pd.Series([], dtype="float64"),
         ),
         (
             pd.DataFrame({"a": [1, 2, 3], "b": [4, 5, 6]}, index=[0, 1, 3]),
             1,
-            pd.Series([]),
+            pd.Series([], dtype="float64"),
         ),
         (pd.Series([1, 2.5, 6]), None, None),
     ],
@@ -1028,7 +1018,6 @@ def test_reductions_non_numeric_dtypes():
 
     # ToDo: pandas supports timedelta std, dask returns float64
     # assert_eq(dds.std(), pds.std())
-    assert_eq(dds.var(), pds.var())
 
     # ToDo: pandas supports timedelta std, otherwise dask raises:
     # TypeError: unsupported operand type(s) for *: 'float' and 'Timedelta'
@@ -1147,13 +1136,23 @@ def test_reductions_frame_dtypes():
     df_no_timedelta = df.drop("timedelta", axis=1, inplace=False)
     ddf_no_timedelta = dd.from_pandas(df_no_timedelta, 3)
 
-    assert_eq(df.sum(), ddf.sum())
+    if not PANDAS_GT_100:
+        # https://github.com/pandas-dev/pandas/issues/30886
+        assert_eq(df.sum(), ddf.sum())
+        assert_eq(df_no_timedelta.mean(), ddf_no_timedelta.mean())
+    else:
+        assert_eq(df.drop(columns="dt").sum(), ddf.drop(columns="dt").sum())
+        assert_eq(
+            df_no_timedelta.drop(columns="dt").mean(),
+            ddf_no_timedelta.drop(columns="dt").mean(),
+        )
+
     assert_eq(df.prod(), ddf.prod())
     assert_eq(df.min(), ddf.min())
     assert_eq(df.max(), ddf.max())
     assert_eq(df.count(), ddf.count())
     assert_eq(df_no_timedelta.std(), ddf_no_timedelta.std())
-    assert_eq(df.var(), ddf.var())
+    assert_eq(df_no_timedelta.var(), ddf_no_timedelta.var())
     if PANDAS_GT_0250:
         # https://github.com/pandas-dev/pandas/issues/18880
         assert_eq(
@@ -1176,8 +1175,6 @@ def test_reductions_frame_dtypes():
         assert_eq(df.var(ddof=0, skipna=False), ddf.var(ddof=0, skipna=False))
     assert_eq(df.sem(ddof=0), ddf.sem(ddof=0))
 
-    assert_eq(df_no_timedelta.mean(), ddf_no_timedelta.mean())
-
     assert_eq(df._get_numeric_data(), ddf._get_numeric_data())
 
     numerics = ddf[["int", "float"]]
diff --git a/dask/dataframe/tests/test_categorical.py b/dask/dataframe/tests/test_categorical.py
index be9ca91d1..07553c9e1 100644
--- a/dask/dataframe/tests/test_categorical.py
+++ b/dask/dataframe/tests/test_categorical.py
@@ -2,11 +2,12 @@ import operator
 
 import numpy as np
 import pandas as pd
-import pandas.util.testing as tm
 import pytest
 
 import dask
 import dask.dataframe as dd
+from dask.dataframe._compat import tm
+from dask.dataframe import _compat
 from dask.dataframe.core import _concat
 from dask.dataframe.utils import (
     make_meta,
@@ -213,7 +214,7 @@ def test_categorize():
 
 def test_categorize_index():
     # Object dtype
-    ddf = dd.from_pandas(tm.makeDataFrame(), npartitions=5)
+    ddf = dd.from_pandas(_compat.makeDataFrame(), npartitions=5)
     df = ddf.compute()
 
     ddf2 = ddf.categorize()
diff --git a/dask/dataframe/tests/test_dataframe.py b/dask/dataframe/tests/test_dataframe.py
index bc72c972b..24f4408a2 100644
--- a/dask/dataframe/tests/test_dataframe.py
+++ b/dask/dataframe/tests/test_dataframe.py
@@ -5,13 +5,14 @@ from operator import add
 import pytest
 import numpy as np
 import pandas as pd
-import pandas.util.testing as tm
 from pandas.io.formats import format as pandas_format
 
 import dask
 import dask.array as da
 from dask.array.numpy_compat import _numpy_118
 import dask.dataframe as dd
+from dask.dataframe import _compat
+from dask.dataframe._compat import tm, PANDAS_GT_0230, PANDAS_GT_100
 from dask.base import compute_as_if_collection
 from dask.utils import put_lines, M
 
@@ -177,7 +178,7 @@ def test_attributes():
     assert "a" in dir(df)
     assert 5 not in dir(df)
 
-    df = dd.from_pandas(tm.makeTimeDataFrame(), npartitions=3)
+    df = dd.from_pandas(_compat.makeTimeDataFrame(), npartitions=3)
     pytest.raises(AttributeError, lambda: df.foo)
 
 
@@ -1542,7 +1543,7 @@ def test_dataframe_picklable():
     cloudpickle = pytest.importorskip("cloudpickle")
     cp_dumps = cloudpickle.dumps
 
-    d = tm.makeTimeDataFrame()
+    d = _compat.makeTimeDataFrame()
     df = dd.from_pandas(d, npartitions=3)
     df = df + 2
 
@@ -1926,7 +1927,7 @@ def test_embarrassingly_parallel_operations():
 
 
 def test_fillna():
-    df = tm.makeMissingDataframe(0.8, 42)
+    df = _compat.makeMissingDataframe()
     ddf = dd.from_pandas(df, npartitions=5, sort=False)
 
     assert_eq(ddf.fillna(100), df.fillna(100))
@@ -1958,7 +1959,8 @@ def test_fillna():
     pytest.raises(NotImplementedError, lambda: ddf.fillna(0, limit=10))
     pytest.raises(NotImplementedError, lambda: ddf.fillna(0, limit=10, axis=1))
 
-    df = tm.makeMissingDataframe(0.2, 42)
+    df = _compat.makeMissingDataframe()
+    df.iloc[:15, 0] = np.nan  # all NaN partition
     ddf = dd.from_pandas(df, npartitions=5, sort=False)
     pytest.raises(ValueError, lambda: ddf.fillna(method="pad").compute())
     assert_eq(df.fillna(method="pad", limit=3), ddf.fillna(method="pad", limit=3))
@@ -1975,7 +1977,7 @@ def test_fillna_duplicate_index():
 
 
 def test_fillna_multi_dataframe():
-    df = tm.makeMissingDataframe(0.8, 42)
+    df = _compat.makeMissingDataframe()
     ddf = dd.from_pandas(df, npartitions=5, sort=False)
 
     assert_eq(ddf.A.fillna(ddf.B), df.A.fillna(df.B))
@@ -1983,7 +1985,7 @@ def test_fillna_multi_dataframe():
 
 
 def test_ffill_bfill():
-    df = tm.makeMissingDataframe(0.8, 42)
+    df = _compat.makeMissingDataframe()
     ddf = dd.from_pandas(df, npartitions=5, sort=False)
 
     assert_eq(ddf.ffill(), df.ffill())
@@ -2115,14 +2117,18 @@ def test_select_dtypes(include, exclude):
 
     tm.assert_series_equal(result.dtypes.value_counts(), expected.dtypes.value_counts())
 
-    if PANDAS_VERSION >= "0.23.0":
-        ctx = pytest.warns(FutureWarning)
-    else:
-        ctx = pytest.warns(None)
-
-    with ctx:
-        tm.assert_series_equal(a.get_ftype_counts(), df.get_ftype_counts())
-        tm.assert_series_equal(result.get_ftype_counts(), expected.get_ftype_counts())
+    if not PANDAS_GT_100:
+        # removed in pandas 1.0
+        if PANDAS_GT_0230:
+            ctx = pytest.warns(FutureWarning)
+        else:
+            ctx = pytest.warns(None)
+
+        with ctx:
+            tm.assert_series_equal(a.get_ftype_counts(), df.get_ftype_counts())
+            tm.assert_series_equal(
+                result.get_ftype_counts(), expected.get_ftype_counts()
+            )
 
 
 def test_deterministic_apply_concat_apply_names():
@@ -2604,7 +2610,7 @@ def test_round():
 
 def test_cov():
     # DataFrame
-    df = pd.util.testing.makeMissingDataframe(0.3, 42)
+    df = _compat.makeMissingDataframe()
     ddf = dd.from_pandas(df, npartitions=6)
 
     res = ddf.cov()
@@ -2646,7 +2652,7 @@ def test_cov():
 
 def test_corr():
     # DataFrame
-    df = pd.util.testing.makeMissingDataframe(0.3, 42)
+    df = _compat.makeMissingDataframe()
     ddf = dd.from_pandas(df, npartitions=6)
 
     res = ddf.corr()
@@ -2694,7 +2700,7 @@ def test_corr():
 def test_corr_same_name():
     # Series with same names (see https://github.com/dask/dask/issues/4906)
 
-    df = pd.util.testing.makeMissingDataframe(0.3, 42)
+    df = _compat.makeMissingDataframe()
     ddf = dd.from_pandas(df, npartitions=6)
 
     result = ddf.A.corr(ddf.B.rename("A"))
@@ -2807,7 +2813,7 @@ def test_apply_infer_columns():
 
 
 def test_index_time_properties():
-    i = tm.makeTimeSeries()
+    i = _compat.makeTimeSeries()
     a = dd.from_pandas(i, npartitions=3)
 
     assert "day" in dir(a.index)
@@ -3027,6 +3033,7 @@ def _assert_info(df, ddf, memory_usage=True):
     stdout_pd = buf_pd.getvalue()
     stdout_da = buf_da.getvalue()
     stdout_da = stdout_da.replace(str(type(ddf)), str(type(df)))
+    # TODO
     assert stdout_pd == stdout_da
 
 
@@ -3137,7 +3144,7 @@ def test_gh_1301():
 
 
 def test_timeseries_sorted():
-    df = tm.makeTimeDataFrame()
+    df = _compat.makeTimeDataFrame()
     ddf = dd.from_pandas(df.reset_index(), npartitions=2)
     df.index.name = "index"
     assert_eq(ddf.set_index("index", sorted=True, drop=True), df)
@@ -3369,7 +3376,7 @@ def test_diff():
 
 
 def test_shift():
-    df = tm.makeTimeDataFrame()
+    df = _compat.makeTimeDataFrame()
     ddf = dd.from_pandas(df, npartitions=4)
 
     # DataFrame
@@ -3392,8 +3399,8 @@ def test_shift():
 
 @pytest.mark.parametrize("data_freq,divs1", [("B", False), ("D", True), ("H", True)])
 def test_shift_with_freq_DatetimeIndex(data_freq, divs1):
-    df = tm.makeTimeDataFrame(30)
-    df = df.set_index(tm.makeDateIndex(30, freq=data_freq))
+    df = _compat.makeTimeDataFrame()
+    df = df.set_index(_compat.makeDateIndex(30, freq=data_freq))
     ddf = dd.from_pandas(df, npartitions=4)
     for freq, divs2 in [("S", True), ("W", False), (pd.Timedelta(10, unit="h"), True)]:
         for d, p in [(ddf, df), (ddf.A, df.A), (ddf.index, df.index)]:
@@ -3408,7 +3415,7 @@ def test_shift_with_freq_DatetimeIndex(data_freq, divs1):
 
 @pytest.mark.parametrize("data_freq,divs", [("B", False), ("D", True), ("H", True)])
 def test_shift_with_freq_PeriodIndex(data_freq, divs):
-    df = tm.makeTimeDataFrame(30)
+    df = _compat.makeTimeDataFrame()
     # PeriodIndex
     df = df.set_index(pd.period_range("2000-01-01", periods=30, freq=data_freq))
     ddf = dd.from_pandas(df, npartitions=4)
@@ -3421,16 +3428,16 @@ def test_shift_with_freq_PeriodIndex(data_freq, divs):
     assert_eq(res, df.index.shift(2))
     assert res.known_divisions == divs
 
-    df = tm.makeTimeDataFrame(30)
+    df = _compat.makeTimeDataFrame()
     with pytest.raises(ValueError):
         ddf.index.shift(2, freq="D")  # freq keyword not supported
 
 
 def test_shift_with_freq_TimedeltaIndex():
-    df = tm.makeTimeDataFrame(30)
+    df = _compat.makeTimeDataFrame()
     # TimedeltaIndex
     for data_freq in ["T", "D", "H"]:
-        df = df.set_index(tm.makeTimedeltaIndex(30, freq=data_freq))
+        df = df.set_index(_compat.makeTimedeltaIndex(30, freq=data_freq))
         ddf = dd.from_pandas(df, npartitions=4)
         for freq in ["S", pd.Timedelta(10, unit="h")]:
             for d, p in [(ddf, df), (ddf.A, df.A), (ddf.index, df.index)]:
@@ -3445,7 +3452,7 @@ def test_shift_with_freq_TimedeltaIndex():
 
 def test_shift_with_freq_errors():
     # Other index types error
-    df = tm.makeDataFrame()
+    df = _compat.makeDataFrame()
     ddf = dd.from_pandas(df, npartitions=4)
     pytest.raises(NotImplementedError, lambda: ddf.shift(2, freq="S"))
     pytest.raises(NotImplementedError, lambda: ddf.A.shift(2, freq="S"))
@@ -3468,9 +3475,20 @@ def test_first_and_last(method):
             assert_eq(f(ddf.A, offset), f(df.A, offset))
 
 
+xfail_hash_object = pytest.mark.xfail(PANDAS_GT_100, reason="GH-30887")
+
+
 @pytest.mark.parametrize("npartitions", [1, 4, 20])
 @pytest.mark.parametrize("split_every", [2, 5])
-@pytest.mark.parametrize("split_out", [None, 1, 5, 20])
+@pytest.mark.parametrize(
+    "split_out",
+    [
+        None,
+        1,
+        pytest.param(5, marks=xfail_hash_object),
+        pytest.param(20, marks=xfail_hash_object),
+    ],
+)
 def test_hash_split_unique(npartitions, split_every, split_out):
     from string import ascii_lowercase
 
@@ -3987,13 +4005,13 @@ def test_scalar_with_array():
 
 def test_has_parallel_type():
     assert has_parallel_type(pd.DataFrame())
-    assert has_parallel_type(pd.Series())
+    assert has_parallel_type(pd.Series(dtype=float))
     assert not has_parallel_type(123)
 
 
 def test_meta_error_message():
     with pytest.raises(TypeError) as info:
-        dd.DataFrame({("x", 1): 123}, "x", pd.Series(), [None, None])
+        dd.DataFrame({("x", 1): 123}, "x", pd.Series(dtype=float), [None, None])
 
     assert "Series" in str(info.value)
     assert "DataFrame" in str(info.value)
@@ -4075,6 +4093,9 @@ def test_dtype_cast():
 @pytest.mark.parametrize("map_npart", [1, 3])
 @pytest.mark.parametrize("sorted_index", [False, True])
 @pytest.mark.parametrize("sorted_map_index", [False, True])
+@pytest.mark.xfail(
+    PANDAS_GT_100, reason="https://github.com/pandas-dev/pandas/issues/30887"
+)
 def test_series_map(base_npart, map_npart, sorted_index, sorted_map_index):
     base = pd.Series(
         ["".join(np.random.choice(["a", "b", "c"], size=3)) for x in range(100)]
@@ -4148,3 +4169,9 @@ def test_iter():
     assert list(df) == list(ddf)
     for col, expected in zip(ddf, ["A", "B"]):
         assert col == expected
+
+
+def test_dataframe_groupby_agg_empty_partitions():
+    df = pd.DataFrame({"x": [1, 2, 3, 4, 5, 6, 7, 8]})
+    ddf = dd.from_pandas(df, npartitions=4)
+    assert_eq(ddf[ddf.x < 5].x.cumsum(), df[df.x < 5].x.cumsum())
diff --git a/dask/dataframe/tests/test_format.py b/dask/dataframe/tests/test_format.py
index 72dd0f548..3ff74a86d 100644
--- a/dask/dataframe/tests/test_format.py
+++ b/dask/dataframe/tests/test_format.py
@@ -527,3 +527,41 @@ def test_duplicate_columns_repr():
     arr = da.from_array(np.arange(10).reshape(5, 2), chunks=(5, 2))
     frame = dd.from_dask_array(arr, columns=["a", "a"])
     repr(frame)
+
+
+def test_empty_repr():
+    df = pd.DataFrame()
+    ddf = dd.from_pandas(df, npartitions=1)
+    exp = (
+        "Empty Dask DataFrame Structure:\n"
+        "Columns: []\n"
+        "Divisions: [, ]\n"
+        "Dask Name: from_pandas, 1 tasks"
+    )
+    assert repr(ddf) == exp
+    exp_table = """<table border="1" class="dataframe">
+  <thead>
+    <tr style="text-align: right;">
+      <th></th>
+    </tr>
+    <tr>
+      <th>npartitions=1</th>
+    </tr>
+  </thead>
+  <tbody>
+    <tr>
+      <th></th>
+    </tr>
+    <tr>
+      <th></th>
+    </tr>
+  </tbody>
+</table>"""
+    exp = """<div><strong>Dask DataFrame Structure:</strong></div>
+<div>
+{style}{exp_table}
+</div>
+<div>Dask Name: from_pandas, 1 tasks</div>""".format(
+        style=style, exp_table=exp_table
+    )
+    assert ddf._repr_html_() == exp
diff --git a/dask/dataframe/tests/test_groupby.py b/dask/dataframe/tests/test_groupby.py
index 5dad375ff..3ef7cf017 100644
--- a/dask/dataframe/tests/test_groupby.py
+++ b/dask/dataframe/tests/test_groupby.py
@@ -4,12 +4,13 @@ from packaging import version
 
 import numpy as np
 import pandas as pd
-import pandas.util.testing as tm
 
 import pytest
 
 import dask
 import dask.dataframe as dd
+from dask.dataframe._compat import tm, PANDAS_GT_100
+from dask.dataframe import _compat
 from dask.dataframe.utils import (
     assert_eq,
     assert_dask_graph,
@@ -475,7 +476,7 @@ def test_series_groupby_errors():
 
 
 def test_groupby_index_array():
-    df = tm.makeTimeDataFrame()
+    df = _compat.makeTimeDataFrame()
     ddf = dd.from_pandas(df, npartitions=2)
 
     # first select column, then group
@@ -494,7 +495,7 @@ def test_groupby_index_array():
 
 
 def test_groupby_set_index():
-    df = tm.makeTimeDataFrame()
+    df = _compat.makeTimeDataFrame()
     ddf = dd.from_pandas(df, npartitions=2)
     pytest.raises(TypeError, lambda: ddf.groupby(df.index.month, as_index=False))
 
@@ -681,7 +682,10 @@ def test_split_apply_combine_on_series(empty):
     pytest.raises(KeyError, lambda: ddf.groupby("x"))
     pytest.raises(KeyError, lambda: ddf.groupby(["a", "x"]))
     pytest.raises(KeyError, lambda: ddf.groupby("a")["x"])
-    pytest.raises(KeyError, lambda: ddf.groupby("a")["b", "x"])
+    with warnings.catch_warnings():
+        # pandas warns about using tuples before throwing the KeyError
+        warnings.simplefilter("ignore", FutureWarning)
+        pytest.raises(KeyError, lambda: ddf.groupby("a")["b", "x"])
     pytest.raises(KeyError, lambda: ddf.groupby("a")[["b", "x"]])
 
     # test graph node labels
@@ -878,7 +882,7 @@ def test_numeric_column_names():
 
 
 def test_groupby_apply_tasks():
-    df = pd.util.testing.makeTimeDataFrame()
+    df = _compat.makeTimeDataFrame()
     df["A"] = df.A // 0.1
     df["B"] = df.B // 0.1
     ddf = dd.from_pandas(df, npartitions=10)
@@ -974,11 +978,13 @@ def test_aggregate__examples(spec, split_every, grouper):
     # Warning from pandas deprecation .agg(dict[dict])
     # it's from pandas, so no reason to assert the deprecation warning,
     # but we should still test it for now
-    with pytest.warns(None):
-        assert_eq(
-            pdf.groupby(grouper(pdf)).agg(spec),
-            ddf.groupby(grouper(ddf)).agg(spec, split_every=split_every),
-        )
+    if not PANDAS_GT_100:
+        # removed in pandas 1.0
+        with pytest.warns(None):
+            assert_eq(
+                pdf.groupby(grouper(pdf)).agg(spec),
+                ddf.groupby(grouper(ddf)).agg(spec, split_every=split_every),
+            )
 
 
 @pytest.mark.parametrize(
@@ -1013,11 +1019,13 @@ def test_series_aggregate__examples(spec, split_every, grouper):
     # Warning from pandas deprecation .agg(dict[dict])
     # it's from pandas, so no reason to assert the deprecation warning,
     # but we should still test it for now
-    with pytest.warns(None):
-        assert_eq(
-            ps.groupby(grouper(pdf)).agg(spec),
-            ds.groupby(grouper(ddf)).agg(spec, split_every=split_every),
-        )
+    if not PANDAS_GT_100:
+        # removed in pandas 1.0
+        with pytest.warns(None):
+            assert_eq(
+                ps.groupby(grouper(pdf)).agg(spec),
+                ds.groupby(grouper(ddf)).agg(spec, split_every=split_every),
+            )
 
 
 def test_aggregate__single_element_groups(agg_func):
@@ -2281,7 +2289,7 @@ def test_groupby_dropna_cudf(dropna, by):
         dask_result = ddf.groupby(by, dropna=dropna).e.sum()
         cudf_result = df.groupby(by, dropna=dropna).e.sum()
     if by in ["c", "d"]:
-        # Loose string/category index name in cudf...
+        # Lose string/category index name in cudf...
         dask_result = dask_result.compute()
         dask_result.index.name = cudf_result.index.name
 
@@ -2296,3 +2304,28 @@ def test_rounding_negative_var():
 
     ddf = dd.from_pandas(df, npartitions=2)
     assert_eq(ddf.groupby("ids").x.std(), df.groupby("ids").x.std())
+
+
+@pytest.mark.parametrize("split_out", [2, 3])
+@pytest.mark.parametrize("column", [["b", "c"], ["b", "d"], ["b", "e"]])
+def test_groupby_split_out_multiindex(split_out, column):
+    df = pd.DataFrame(
+        {
+            "a": np.arange(8),
+            "b": [1, 0, 0, 2, 1, 1, 2, 0],
+            "c": [0, 1] * 4,
+            "d": ["dog", "cat", "cat", "dog", "dog", "dog", "cat", "bird"],
+        }
+    ).fillna(0)
+    df["e"] = df["d"].astype("category")
+    ddf = dd.from_pandas(df, npartitions=3)
+
+    ddf_result_so1 = (
+        ddf.groupby(column).a.mean(split_out=1).compute().sort_values().dropna()
+    )
+
+    ddf_result = (
+        ddf.groupby(column).a.mean(split_out=split_out).compute().sort_values().dropna()
+    )
+
+    assert_eq(ddf_result, ddf_result_so1, check_index=False)
diff --git a/dask/dataframe/tests/test_hashing.py b/dask/dataframe/tests/test_hashing.py
index ca37bbb1a..d046efb62 100644
--- a/dask/dataframe/tests/test_hashing.py
+++ b/dask/dataframe/tests/test_hashing.py
@@ -1,7 +1,8 @@
 import numpy as np
 import pandas as pd
-import pandas.util.testing as tm
 import dask.dataframe as dd
+from dask.dataframe import _compat
+from dask.dataframe._compat import tm, PANDAS_GT_100
 from pandas.util import hash_pandas_object
 
 import pytest
@@ -20,11 +21,11 @@ from dask.dataframe.utils import assert_eq
         pd.Index([1, 2, 3]),
         pd.Index([True, False, True]),
         pd.DataFrame({"x": ["a", "b", "c"], "y": [1, 2, 3]}),
-        pd.util.testing.makeMissingDataframe(),
-        pd.util.testing.makeMixedDataFrame(),
-        pd.util.testing.makeTimeDataFrame(),
-        pd.util.testing.makeTimeSeries(),
-        pd.util.testing.makeTimedeltaIndex(),
+        _compat.makeMissingDataframe(),
+        _compat.makeMixedDataFrame(),
+        _compat.makeTimeDataFrame(),
+        _compat.makeTimeSeries(),
+        _compat.makeTimedeltaIndex(),
     ],
 )
 def test_hash_pandas_object(obj):
@@ -64,16 +65,24 @@ def test_object_missing_values():
     tm.assert_series_equal(h1, h2)
 
 
+xfail_hash_object = pytest.mark.xfail(
+    PANDAS_GT_100, reason="https://github.com/pandas-dev/pandas/issues/30887"
+)
+
+
 @pytest.mark.parametrize(
     "obj",
     [
         pd.Index([1, 2, 3]),
-        pd.Index([True, False, True], index=[1.5, 1.1, 3.3]),
+        pytest.param(pd.Index([True, False, True]), marks=xfail_hash_object),
         pd.Series([1, 2, 3]),
         pd.Series([1.0, 1.5, 3.2]),
         pd.Series([1.0, 1.5, 3.2], index=[1.5, 1.1, 3.3]),
         pd.DataFrame({"x": ["a", "b", "c"], "y": [1, 2, 3]}),
-        pd.DataFrame({"x": ["a", "b", "c"], "y": [1, 2, 3]}, index=["a", "z", "x"]),
+        pytest.param(
+            pd.DataFrame({"x": ["a", "b", "c"], "y": [1, 2, 3]}, index=["a", "z", "x"]),
+            marks=xfail_hash_object,
+        ),
     ],
 )
 def test_hash_object_dispatch(obj):
diff --git a/dask/dataframe/tests/test_indexing.py b/dask/dataframe/tests/test_indexing.py
index 68cf5b7e7..3c10de580 100644
--- a/dask/dataframe/tests/test_indexing.py
+++ b/dask/dataframe/tests/test_indexing.py
@@ -1,11 +1,11 @@
 import pandas as pd
-import pandas.util.testing as tm
 import numpy as np
 
 import pytest
 
 import dask.dataframe as dd
 
+from dask.dataframe._compat import tm, PANDAS_GT_100
 from dask.dataframe.indexing import _coerce_loc_index
 from dask.dataframe.utils import assert_eq, make_meta, PANDAS_VERSION
 
@@ -37,24 +37,30 @@ def test_loc():
     else:
         expected_warning = None
 
-    with pytest.warns(expected_warning):
-        assert_eq(d.loc[[3, 4, 1, 8]], full.loc[[3, 4, 1, 8]])
-    with pytest.warns(expected_warning):
-        assert_eq(d.loc[[3, 4, 1, 9]], full.loc[[3, 4, 1, 9]])
-    with pytest.warns(expected_warning):
-        assert_eq(d.loc[np.array([3, 4, 1, 9])], full.loc[np.array([3, 4, 1, 9])])
+    if not PANDAS_GT_100:
+        # removed in pandas 1.0
+        with pytest.warns(expected_warning):
+            assert_eq(d.loc[[3, 4, 1, 8]], full.loc[[3, 4, 1, 8]])
+        with pytest.warns(expected_warning):
+            assert_eq(d.loc[[3, 4, 1, 9]], full.loc[[3, 4, 1, 9]])
+        with pytest.warns(expected_warning):
+            assert_eq(d.loc[np.array([3, 4, 1, 9])], full.loc[np.array([3, 4, 1, 9])])
 
     assert_eq(d.a.loc[5], full.a.loc[5:5])
     assert_eq(d.a.loc[3:8], full.a.loc[3:8])
     assert_eq(d.a.loc[:8], full.a.loc[:8])
     assert_eq(d.a.loc[3:], full.a.loc[3:])
     assert_eq(d.a.loc[[5]], full.a.loc[[5]])
-    with pytest.warns(expected_warning):
-        assert_eq(d.a.loc[[3, 4, 1, 8]], full.a.loc[[3, 4, 1, 8]])
-    with pytest.warns(expected_warning):
-        assert_eq(d.a.loc[[3, 4, 1, 9]], full.a.loc[[3, 4, 1, 9]])
-    with pytest.warns(expected_warning):
-        assert_eq(d.a.loc[np.array([3, 4, 1, 9])], full.a.loc[np.array([3, 4, 1, 9])])
+    if not PANDAS_GT_100:
+        # removed in pandas 1.0
+        with pytest.warns(expected_warning):
+            assert_eq(d.a.loc[[3, 4, 1, 8]], full.a.loc[[3, 4, 1, 8]])
+        with pytest.warns(expected_warning):
+            assert_eq(d.a.loc[[3, 4, 1, 9]], full.a.loc[[3, 4, 1, 9]])
+        with pytest.warns(expected_warning):
+            assert_eq(
+                d.a.loc[np.array([3, 4, 1, 9])], full.a.loc[np.array([3, 4, 1, 9])]
+            )
     assert_eq(d.a.loc[[]], full.a.loc[[]])
     assert_eq(d.a.loc[np.array([])], full.a.loc[np.array([])])
 
@@ -82,8 +88,8 @@ def test_loc_non_informative_index():
 
 
 def test_loc_with_text_dates():
-    A = tm.makeTimeSeries(10).iloc[:5]
-    B = tm.makeTimeSeries(10).iloc[5:]
+    A = dd._compat.makeTimeSeries().iloc[:5]
+    B = dd._compat.makeTimeSeries().iloc[5:]
     s = dd.Series(
         {("df", 0): A, ("df", 1): B},
         "df",
@@ -168,6 +174,7 @@ def test_loc2d():
         d.a.loc[d.a % 2 == 0, 3]
 
 
+@pytest.mark.skip(PANDAS_GT_100, reason="Removed in pandas 1.0")
 def test_loc2d_some_missing():
     with pytest.warns(FutureWarning):
         assert_eq(d.loc[[3, 4, 3], ["a"]], full.loc[[3, 4, 3], ["a"]])
diff --git a/dask/dataframe/tests/test_multi.py b/dask/dataframe/tests/test_multi.py
index 34e67f6a6..1e1957c6a 100644
--- a/dask/dataframe/tests/test_multi.py
+++ b/dask/dataframe/tests/test_multi.py
@@ -4,9 +4,9 @@ import dask
 import dask.dataframe as dd
 import numpy as np
 import pandas as pd
-import pandas.util.testing as tm
 
 from dask.base import compute_as_if_collection
+from dask.dataframe._compat import tm, PANDAS_GT_100
 from dask.dataframe.core import _Frame
 from dask.dataframe.methods import concat, concat_kwargs
 from dask.dataframe.multi import (
@@ -221,7 +221,8 @@ def list_eq(aa, bb):
     else:
         av = a.sort_values().values
         bv = b.sort_values().values
-    tm.assert_numpy_array_equal(av, bv)
+
+    dd._compat.assert_numpy_array_equal(av, bv)
 
 
 @pytest.mark.parametrize("how", ["inner", "left", "right", "outer"])
@@ -1822,20 +1823,10 @@ def test_append2():
 
     assert_eq(ddf1.append(ddf2), ddf1.compute().append(ddf2.compute(), **concat_kwargs))
     assert_eq(ddf2.append(ddf1), ddf2.compute().append(ddf1.compute(), **concat_kwargs))
-    # Series + DataFrame
-    with pytest.warns(None):
-        # RuntimeWarning from pandas on comparing int and str
-        assert_eq(ddf1.a.append(ddf2), ddf1.a.compute().append(ddf2.compute()))
-        assert_eq(ddf2.a.append(ddf1), ddf2.a.compute().append(ddf1.compute()))
 
     # different columns
     assert_eq(ddf1.append(ddf3), ddf1.compute().append(ddf3.compute(), **concat_kwargs))
     assert_eq(ddf3.append(ddf1), ddf3.compute().append(ddf1.compute(), **concat_kwargs))
-    # Series + DataFrame
-    with pytest.warns(None):
-        # RuntimeWarning from pandas on comparing int and str
-        assert_eq(ddf1.a.append(ddf3), ddf1.a.compute().append(ddf3.compute()))
-        assert_eq(ddf3.b.append(ddf1), ddf3.b.compute().append(ddf1.compute()))
 
     # Dask + pandas
     assert_eq(
@@ -1846,15 +1837,6 @@ def test_append2():
         ddf2.append(ddf1.compute()),
         ddf2.compute().append(ddf1.compute(), **concat_kwargs),
     )
-    # Series + DataFrame
-    with pytest.warns(None):
-        # RuntimeWarning from pandas on comparing int and str
-        assert_eq(
-            ddf1.a.append(ddf2.compute()), ddf1.a.compute().append(ddf2.compute())
-        )
-        assert_eq(
-            ddf2.a.append(ddf1.compute()), ddf2.a.compute().append(ddf1.compute())
-        )
 
     # different columns
     assert_eq(
@@ -1865,15 +1847,6 @@ def test_append2():
         ddf3.append(ddf1.compute()),
         ddf3.compute().append(ddf1.compute(), **concat_kwargs),
     )
-    # Series + DataFrame
-    with pytest.warns(None):
-        # RuntimeWarning from pandas on comparing int and str
-        assert_eq(
-            ddf1.a.append(ddf3.compute()), ddf1.a.compute().append(ddf3.compute())
-        )
-        assert_eq(
-            ddf3.b.append(ddf1.compute()), ddf3.b.compute().append(ddf1.compute())
-        )
 
 
 def test_append_categorical():
@@ -1971,6 +1944,9 @@ def test_multi_duplicate_divisions():
     assert_eq(r1, r2)
 
 
+@pytest.mark.xfail(
+    PANDAS_GT_100, reason="https://github.com/pandas-dev/pandas/issues/30887"
+)
 def test_merge_outer_empty():
     # Issue #5470 bug reproducer
     k_clusters = 3
diff --git a/dask/dataframe/tests/test_reshape.py b/dask/dataframe/tests/test_reshape.py
index 910cdccfd..829663ef4 100644
--- a/dask/dataframe/tests/test_reshape.py
+++ b/dask/dataframe/tests/test_reshape.py
@@ -1,10 +1,10 @@
 import numpy as np
 import pandas as pd
-import pandas.util.testing as tm
 import pytest
 
 import dask.dataframe as dd
 
+from dask.dataframe._compat import tm
 from dask.dataframe.utils import assert_eq, make_meta, PANDAS_VERSION, PANDAS_GT_0240
 
 
diff --git a/dask/dataframe/tests/test_shuffle.py b/dask/dataframe/tests/test_shuffle.py
index f80ae0b35..91ecb4087 100644
--- a/dask/dataframe/tests/test_shuffle.py
+++ b/dask/dataframe/tests/test_shuffle.py
@@ -9,10 +9,10 @@ import numpy as np
 import string
 import multiprocessing as mp
 from copy import copy
-import pandas.util.testing as tm
 
 import dask
 import dask.dataframe as dd
+from dask.dataframe._compat import tm, PANDAS_GT_100, assert_categorical_equal
 from dask import delayed
 from dask.base import compute_as_if_collection
 from dask.dataframe.shuffle import (
@@ -142,6 +142,9 @@ def test_partitioning_index():
     assert len(np.unique(res)) > 1
 
 
+@pytest.mark.xfail(
+    PANDAS_GT_100, reason="https://github.com/pandas-dev/pandas/issues/30887"
+)
 def test_partitioning_index_categorical_on_values():
     df = pd.DataFrame({"a": list(string.ascii_letters), "b": [1, 2, 3, 4] * 13})
     df.a = df.a.astype("category")
@@ -752,7 +755,7 @@ def test_set_index_categorical():
 
     # sorted with the metric defined by the Categorical
     divisions = pd.Categorical(result.divisions, dtype=dtype)
-    tm.assert_categorical_equal(divisions, divisions.sort_values())
+    assert_categorical_equal(divisions, divisions.sort_values())
 
 
 def test_compute_divisions():
diff --git a/dask/dataframe/tests/test_ufunc.py b/dask/dataframe/tests/test_ufunc.py
index 74c6de606..a74bddaa4 100644
--- a/dask/dataframe/tests/test_ufunc.py
+++ b/dask/dataframe/tests/test_ufunc.py
@@ -1,7 +1,6 @@
 import pytest
 
 pd = pytest.importorskip("pandas")
-import pandas.util.testing as tm
 
 import numpy as np
 
@@ -183,10 +182,10 @@ def test_ufunc_array_wrap(ufunc):
     assert_eq(dafunc(ds), pd.Series(npfunc(s), index=s.index))
 
     assert isinstance(npfunc(ds), np.ndarray)
-    tm.assert_numpy_array_equal(npfunc(ds), npfunc(s))
+    np.testing.assert_equal(npfunc(ds), npfunc(s))
 
     assert isinstance(dafunc(s), np.ndarray)
-    tm.assert_numpy_array_equal(dafunc(s), npfunc(s))
+    np.testing.assert_array_equal(dafunc(s), npfunc(s))
 
     df = pd.DataFrame(
         {
@@ -205,10 +204,10 @@ def test_ufunc_array_wrap(ufunc):
     assert_eq(dafunc(ddf), exp)
 
     assert isinstance(npfunc(ddf), np.ndarray)
-    tm.assert_numpy_array_equal(npfunc(ddf), npfunc(df))
+    np.testing.assert_array_equal(npfunc(ddf), npfunc(df))
 
     assert isinstance(dafunc(df), np.ndarray)
-    tm.assert_numpy_array_equal(dafunc(df), npfunc(df))
+    np.testing.assert_array_equal(dafunc(df), npfunc(df))
 
 
 _UFUNCS_2ARG = [
@@ -453,7 +452,7 @@ def test_2args_with_array(ufunc, pandas, darray):
     assert isinstance(dafunc(dask, darray), dask_type)
     assert isinstance(dafunc(darray, dask), dask_type)
 
-    tm.assert_numpy_array_equal(
+    np.testing.assert_array_equal(
         dafunc(dask, darray).compute().values, npfunc(pandas.values, darray).compute()
     )
 
@@ -461,10 +460,10 @@ def test_2args_with_array(ufunc, pandas, darray):
     assert isinstance(npfunc(dask, darray), dask_type)
     assert isinstance(npfunc(darray, dask), dask_type)
 
-    tm.assert_numpy_array_equal(
+    np.testing.assert_array_equal(
         npfunc(dask, darray).compute().values, npfunc(pandas.values, darray.compute())
     )
-    tm.assert_numpy_array_equal(
+    np.testing.assert_array_equal(
         npfunc(darray, dask).compute().values, npfunc(darray.compute(), pandas.values)
     )
 
diff --git a/dask/dataframe/tests/test_utils_dataframe.py b/dask/dataframe/tests/test_utils_dataframe.py
index 764a18251..103cc1b9f 100644
--- a/dask/dataframe/tests/test_utils_dataframe.py
+++ b/dask/dataframe/tests/test_utils_dataframe.py
@@ -2,8 +2,8 @@ import re
 
 import numpy as np
 import pandas as pd
-import pandas.util.testing as tm
 import dask.dataframe as dd
+from dask.dataframe._compat import tm
 from dask.dataframe.core import apply_and_enforce
 from dask.dataframe.utils import (
     shard_df_on_index,
diff --git a/dask/tests/test_base.py b/dask/tests/test_base.py
index c6b2ddffd..7f2f17cb0 100644
--- a/dask/tests/test_base.py
+++ b/dask/tests/test_base.py
@@ -617,8 +617,8 @@ def test_compute_dataframe():
     ddf1 = ddf.a + 1
     ddf2 = ddf.a + ddf.b
     out1, out2 = compute(ddf1, ddf2)
-    pd.util.testing.assert_series_equal(out1, df.a + 1)
-    pd.util.testing.assert_series_equal(out2, df.a + df.b)
+    pd.testing.assert_series_equal(out1, df.a + 1)
+    pd.testing.assert_series_equal(out2, df.a + df.b)
 
 
 @pytest.mark.skipif("not dd or not da")
@@ -629,7 +629,7 @@ def test_compute_array_dataframe():
     ddf = dd.from_pandas(df, npartitions=2).a + 2
     arr_out, df_out = compute(darr, ddf)
     assert np.allclose(arr_out, arr + 1)
-    pd.util.testing.assert_series_equal(df_out, df.a + 2)
+    dd._compat.tm.assert_series_equal(df_out, df.a + 2)
 
 
 @pytest.mark.skipif("not dd")
diff --git a/dask/tests/test_distributed.py b/dask/tests/test_distributed.py
index 4bd3a25d9..2ad52aa82 100644
--- a/dask/tests/test_distributed.py
+++ b/dask/tests/test_distributed.py
@@ -136,8 +136,12 @@ def test_to_hdf_distributed(c):
     ],
 )
 def test_to_hdf_scheduler_distributed(npartitions, c):
+    dd = pytest.importorskip("dask.dataframe")
     from ..dataframe.io.tests.test_hdf import test_to_hdf_schedulers
 
+    if dd._compat.PANDAS_GT_100:
+        pytest.skip("https://github.com/pandas-dev/pandas/issues/30962")
+
     test_to_hdf_schedulers(None, npartitions)
 
 
diff --git a/dask/tests/test_order.py b/dask/tests/test_order.py
index cef23e0f1..98e450f48 100644
--- a/dask/tests/test_order.py
+++ b/dask/tests/test_order.py
@@ -53,9 +53,7 @@ def test_avoid_broker_nodes(abcde):
         (b, 1): (f, (a, 1)),
         (b, 2): (f, (a, 1)),
     }
-
     o = order(dsk)
-
     assert o[(a, 0)] < o[(a, 1)]
 
     # Switch name of 0, 1 to ensure that this isn't due to string comparison
@@ -66,11 +64,20 @@ def test_avoid_broker_nodes(abcde):
         (b, 1): (f, (a, 0)),
         (b, 2): (f, (a, 0)),
     }
-
     o = order(dsk)
-
     assert o[(a, 0)] > o[(a, 1)]
 
+    # Switch name of 0, 1 for "b"s too
+    dsk = {
+        (a, 0): (f,),
+        (a, 1): (f,),
+        (b, 1): (f, (a, 0)),
+        (b, 0): (f, (a, 1)),
+        (b, 2): (f, (a, 1)),
+    }
+    o = order(dsk)
+    assert o[(a, 0)] < o[(a, 1)]
+
 
 def test_base_of_reduce_preferred(abcde):
     r"""
@@ -168,11 +175,10 @@ def test_avoid_upwards_branching_complex(abcde):
     }
 
     o = order(dsk)
-
     assert o[(c, 1)] < o[(b, 1)]
+    assert abs(o[(d, 2)] - o[(d, 3)]) == 1
 
 
-@pytest.mark.xfail(reason="this case is ambiguous", strict=False)
 def test_deep_bases_win_over_dependents(abcde):
     r"""
     It's not clear who should run first, e or d
@@ -181,6 +187,8 @@ def test_deep_bases_win_over_dependents(abcde):
     2.  e is nicer (hypothetically) because it will be sooner released
         (though in this case we need d to run first regardless)
 
+    Regardless of e or d first, we should run b before c.
+
             a
           / | \   .
          b  c |
@@ -191,8 +199,8 @@ def test_deep_bases_win_over_dependents(abcde):
     dsk = {a: (f, b, c, d), b: (f, d, e), c: (f, d), d: 1, e: 2}
 
     o = order(dsk)
-    assert o[e] < o[d]
-    assert o[d] < o[b] or o[d] < o[c]
+    assert o[e] < o[d]  # ambiguous, but this is what we currently expect
+    assert o[b] < o[c]
 
 
 def test_prefer_deep(abcde):
@@ -219,7 +227,6 @@ def test_stacklimit(abcde):
     ndependencies(dependencies, dependents)
 
 
-@pytest.mark.xfail(reason="Can't please 'em all")
 def test_break_ties_by_str(abcde):
     a, b, c, d, e = abcde
     dsk = {("x", i): (inc, i) for i in range(10)}
@@ -227,8 +234,8 @@ def test_break_ties_by_str(abcde):
     dsk["y"] = list(x_keys)
 
     o = order(dsk)
-    expected = {"y": 0}
-    expected.update({k: i + 1 for i, k in enumerate(x_keys)})
+    expected = {"y": 10}
+    expected.update({k: i for i, k in enumerate(x_keys)})
 
     assert o == expected
 
@@ -494,6 +501,7 @@ def test_prefer_short_ancestor(abcde):
     }
     o = order(dsk)
 
+    assert o[(a, 0)] < o[(a, 1)]
     assert o[(b, 0)] < o[(b, 1)]
     assert o[(b, 0)] < o[(c, 2)]
     assert o[(c, 1)] < o[(c, 2)]
@@ -535,3 +543,116 @@ def test_map_overlap(abcde):
     o = order(dsk)
 
     assert o[(b, 1)] < o[(e, 5)] or o[(b, 5)] < o[(e, 1)]
+
+
+def test_use_structure_not_keys(abcde):
+    """ See https://github.com/dask/dask/issues/5584#issuecomment-554963958
+
+    We were using key names to infer structure, which could result in funny behavior.
+    """
+    a, b, _, _, _ = abcde
+    dsk = {
+        (a, 0): (f,),
+        (a, 1): (f,),
+        (a, 2): (f,),
+        (a, 3): (f,),
+        (a, 4): (f,),
+        (a, 5): (f,),
+        (a, 6): (f,),
+        (a, 7): (f,),
+        (a, 8): (f,),
+        (a, 9): (f,),
+        (b, 5): (f, (a, 2)),
+        (b, 7): (f, (a, 0), (a, 2)),
+        (b, 9): (f, (a, 7), (a, 0), (a, 2)),
+        (b, 1): (f, (a, 4), (a, 7), (a, 0)),
+        (b, 2): (f, (a, 9), (a, 4), (a, 7)),
+        (b, 4): (f, (a, 6), (a, 9), (a, 4)),
+        (b, 3): (f, (a, 5), (a, 6), (a, 9)),
+        (b, 8): (f, (a, 1), (a, 5), (a, 6)),
+        (b, 6): (f, (a, 8), (a, 1), (a, 5)),
+        (b, 0): (f, (a, 3), (a, 8), (a, 1)),
+    }
+    o = order(dsk)
+    As = sorted(val for (letter, _), val in o.items() if letter == a)
+    Bs = sorted(val for (letter, _), val in o.items() if letter == b)
+    assert Bs[0] in {1, 3}
+    if Bs[0] == 3:
+        assert As == [0, 1, 2, 4, 6, 8, 10, 12, 14, 16]
+        assert Bs == [3, 5, 7, 9, 11, 13, 15, 17, 18, 19]
+    else:
+        assert As == [0, 2, 4, 6, 8, 10, 12, 14, 16, 18]
+        assert Bs == [1, 3, 5, 7, 9, 11, 13, 15, 17, 19]
+
+
+def test_dont_run_all_dependents_too_early(abcde):
+    """ From https://github.com/dask/dask-ml/issues/206#issuecomment-395873372 """
+    a, b, c, d, e = abcde
+    depth = 10
+    dsk = {(a, 0): 0, (b, 0): 1, (c, 0): 2, (d, 0): (f, (a, 0), (b, 0), (c, 0))}
+    for i in range(1, depth):
+        dsk[(b, i)] = (f, (b, 0))
+        dsk[(c, i)] = (f, (c, 0))
+        dsk[(d, i)] = (f, (d, i - 1), (b, i), (c, i))
+    o = order(dsk)
+    expected = [3, 6, 9, 12, 15, 18, 21, 24, 27, 30]
+    actual = sorted(v for (letter, num), v in o.items() if letter == d)
+    assert expected == actual
+
+
+def test_many_branches_use_ndependencies(abcde):
+    """ From https://github.com/dask/dask/pull/5646#issuecomment-562700533
+
+    Sometimes we need larger or wider DAGs to test behavior.  This test
+    ensures we choose the branch with more work twice in successtion.
+    This is important, because ``order`` may search along dependencies
+    and then along dependents.
+
+    """
+    a, b, c, d, e = abcde
+    dd = d + d
+    ee = e + e
+    dsk = {
+        (a, 0): 0,
+        (a, 1): (f, (a, 0)),
+        (a, 2): (f, (a, 1)),
+        (b, 1): (f, (a, 0)),
+        (b, 2): (f, (b, 1)),
+        (c, 1): (f, (a, 0)),  # most short and thin; should go last
+        (d, 1): (f, (a, 0)),
+        (d, 2): (f, (d, 1)),
+        (dd, 1): (f, (a, 0)),
+        (dd, 2): (f, (dd, 1)),
+        (dd, 3): (f, (d, 2), (dd, 2)),
+        (e, 1): (f, (a, 0)),
+        (e, 2): (f, (e, 1)),
+        (ee, 1): (f, (a, 0)),
+        (ee, 2): (f, (ee, 1)),
+        (ee, 3): (f, (e, 2), (ee, 2)),
+        (a, 3): (f, (a, 2), (b, 2), (c, 1), (dd, 3), (ee, 3)),
+    }
+    o = order(dsk)
+    # run all d's and e's first
+    expected = [1, 2, 3, 4, 5, 6, 7, 8, 9, 10]
+    actual = sorted(v for (letter, _), v in o.items() if letter in {d, dd, e, ee})
+    assert actual == expected
+    assert o[(c, 1)] == o[(a, 3)] - 1
+
+
+def test_order_cycle():
+    with pytest.raises(RuntimeError, match="Cycle detected"):
+        dask.get({"a": (f, "a")}, "a")  # we encounter this in `get`
+    with pytest.raises(RuntimeError, match="Cycle detected"):
+        order({"a": (f, "a")})  # trivial self-loop
+    with pytest.raises(RuntimeError, match="Cycle detected"):
+        order({("a", 0): (f, ("a", 0))})  # non-string
+    with pytest.raises(RuntimeError, match="Cycle detected"):
+        order({"a": (f, "b"), "b": (f, "c"), "c": (f, "a")})  # non-trivial loop
+    with pytest.raises(RuntimeError, match="Cycle detected"):
+        order({"a": (f, "b"), "b": (f, "c"), "c": (f, "a", "d"), "d": 1})
+    with pytest.raises(RuntimeError, match="Cycle detected"):
+        order({"a": (f, "b"), "b": (f, "c"), "c": (f, "a", "d"), "d": (f, "b")})
+
+
+def test_order_empty():
+    assert order({}) == {}