Repository: dask/dask

Issue:
Possible bug when checking an empty dataframe
I am aware of [this question][1]. But check the code(minimal-working example) below:
```
import dask.dataframe as dd
import pandas as pd

# intialise data of lists.
data = {'Name': ['Tom', 'nick', 'krish', 'jack'], 'Age': [20, 21, 19, 18]}

# Create DataFrame
df = pd.DataFrame(data)
dask_df = dd.from_pandas(df, npartitions=1)

categoric_df = dask_df.select_dtypes(include="category")
```
When I try to print the `categoric_df` I get the following error:
```
ValueError: No objects to concatenate
```
And when I check the `categoric_df` from PyCharm debugger:

    Unable to get repr for <class 'dask.dataframe.core.DataFrame'>
With these errors, I can build a try/except block to check if the dataframe is empty or not. But I don't want to use this approach since it is not guaranteed to work all the time and try/except slows down the code.
And when I try to print computed `categoric_df` it looks like this:
```
>>>print(categoric_df.compute())
Empty DataFrame
Columns: []
Index: [0, 1, 2, 3]
```
In summary: Here if I select the non-existing dypes and create a dask.DataFrame from it, I get a `dask.DataFrame` which at first glance doesn't seem empty if I use `len()` function.
```
>>>print(len(categoric_df))
4
>>>print(len(categoric_df.compute())
4
>>>print(categoric_df.compute().empty)
True
```
 **Is this a bug?**  Is there a way to check if the `categoric_df` is empty or not without computing it? (I want it to stay lazily evaluated.)

> **UPDATE**: `print(len(categoric_df.columns))` is returning `0`. This can be used for figuring out if the dataframe is empty or not. But is this viable? I am not sure.

  [1]: https://stackoverflow.com/questions/50206730/how-to-check-if-dask-dataframe-is-empty




Discussion:

Patch:
diff --git a/dask/dataframe/core.py b/dask/dataframe/core.py
index cccd4307a..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),
@@ -4297,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}
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/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.


Test patch:
diff --git a/dask/dataframe/io/tests/test_csv.py b/dask/dataframe/io/tests/test_csv.py
index 5f6dbacfe..fbf0bec12 100644
--- a/dask/dataframe/io/tests/test_csv.py
+++ b/dask/dataframe/io/tests/test_csv.py
@@ -672,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_hdf.py b/dask/dataframe/io/tests/test_hdf.py
index 949b4ed4a..0325860fb 100644
--- a/dask/dataframe/io/tests/test_hdf.py
+++ b/dask/dataframe/io/tests/test_hdf.py
@@ -13,6 +13,11 @@ 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 6f5fe0462..e591fd11b 100644
--- a/dask/dataframe/io/tests/test_io.py
+++ b/dask/dataframe/io/tests/test_io.py
@@ -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 be6abd680..0a436b1c0 100644
--- a/dask/dataframe/io/tests/test_parquet.py
+++ b/dask/dataframe/io/tests/test_parquet.py
@@ -9,7 +9,6 @@ import pytest
 import dask
 import dask.multiprocessing
 import dask.dataframe as dd
-from dask.dataframe._compat import tm
 from dask.dataframe.utils import assert_eq, PANDAS_VERSION
 from dask.dataframe.io.parquet.utils import _parse_pandas_metadata
 from dask.dataframe.optimize import optimize_read_parquet_getitem
@@ -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_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_multi.py b/dask/dataframe/tests/test_multi.py
index e79e98c60..1e1957c6a 100644
--- a/dask/dataframe/tests/test_multi.py
+++ b/dask/dataframe/tests/test_multi.py
@@ -1823,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(
@@ -1847,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(
@@ -1866,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():
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({}) == {}