Repository: dask/dask

Issue:
Use shallow copies when assigning columns
Currently when we run code like `df["z"] = df.x + df.y` we call `df = df.assign(z=df.x + df.y)`.  However, the latter snippet calls the underlying assign method, which calls `df.copy()` which can be slow.  Instead, it might be nice to do soemthing like the following:

```python
def set_column(df, name, value):
    df = df.copy(deep=False)
    df[name] = value
    return df
```

And then replace the calls to `M.assign` with this.

This is particularly important for cudf, but probably helps pandas performance a bit as well.

We might want to do this for all assign calls, rather than just setitem?  There are some semantic differences in pandas apparently with regards to copying, but because we don't mutate my guess is that we should do this everywhere.  @TomAugspurger ?

Discussion:
rjzamora: **Note**:  It appears that the current implementation is problematic for `test_attribute_assignment`:

```python
import pandas as pd
import dask.dataframe as dd

df = pd.DataFrame({"x": [1, 2, 3, 4, 5], "y": [1.0, 2.0, 3.0, 4.0, 5.0]})
ddf = dd.from_pandas(df, npartitions=2)
print("Before ddf Modified...\n", df.y)

ddf.y = ddf.x + ddf.y
ddf.persist()
print("\nAfter ddf Modified...\n", df.y)
```

Output:
```
Before ddf Modified...
0    1.0
1    2.0
2    3.0
3    4.0
4    5.0
Name: y, dtype: float64

After ddf Modified...
0     2.0
1     4.0
2     6.0
3     8.0
4    10.0
Name: y, dtype: float64
```
TomAugspurger: Does that issue only show up with attribute assignment or is it present with setitem too? Pandas strongly recommends against assigning to an attribute. 
mrocklin: > Does that issue only show up with attribute assignment or is it present with setitem too? Pandas strongly recommends against assigning to an attribute.

I don't think that we're using setattr internally.  I think that it's more likely that this is what you were referring to earlier, that a shallow copy might not be enough to sufficiently protect the input dataframe.
TomAugspurger: @rjzamora here's a potential workaround: only do a shallow copy when we're adding a new column:

```diff
diff --git a/dask/dataframe/methods.py b/dask/dataframe/methods.py
index e3e393093..e09d8297d 100644
--- a/dask/dataframe/methods.py
+++ b/dask/dataframe/methods.py
@@ -263,8 +263,13 @@ def set_column(df, name, value):
 
 
 def assign(df, *pairs):
-    df = df.copy(deep=False)
-    for name, val in dict(partition(2, pairs)).items():
+    pairs = dict(partition(2, pairs))
+    # deep copy when updating an element, to avoid issues
+    # with modifying the original
+    deep = bool(set(pairs) & set(df.columns))
+
+    df = df.copy(deep=bool(deep))
+    for name, val in pairs.items():
         df[name] = val
     return df
```

I think that avoids the behavior failing the test. Can you check if it retains the performance boost? Do people think this is likely to cause issues elsewhere? I'm having trouble reasoning through things just in my head.
rjzamora: Thanks @TomAugspurger!  I will give it a try.  You have good timing - I was just about to revisit this :) 

Patch:
diff --git a/continuous_integration/appveyor/environment.yaml b/continuous_integration/appveyor/environment.yaml
index 7c1dd0b63..6b0102296 100644
--- a/continuous_integration/appveyor/environment.yaml
+++ b/continuous_integration/appveyor/environment.yaml
@@ -10,6 +10,7 @@ dependencies:
   - pytest
   - cloudpickle
   - distributed
+  - msgpack-python==0.6.2
   - toolz
   - bcolz
   - bokeh
diff --git a/continuous_integration/travis/travis-36.yaml b/continuous_integration/travis/travis-36.yaml
index 6a07d576f..e21d2c8d6 100644
--- a/continuous_integration/travis/travis-36.yaml
+++ b/continuous_integration/travis/travis-36.yaml
@@ -35,6 +35,7 @@ dependencies:
   - crick
   - cytoolz
   - distributed==2.0.1
+  - msgpack-python==0.6.2
   - graphviz
   - ipython
   - lz4
diff --git a/continuous_integration/travis/travis-37.yaml b/continuous_integration/travis/travis-37.yaml
index d7921b6a0..4a134554d 100644
--- a/continuous_integration/travis/travis-37.yaml
+++ b/continuous_integration/travis/travis-37.yaml
@@ -33,6 +33,7 @@ dependencies:
   - crick
   - cytoolz
   - distributed==2.0.1
+  - msgpack-python==0.6.2
   - graphviz
   - ipython
   - lz4
diff --git a/continuous_integration/travis/travis-38.yaml b/continuous_integration/travis/travis-38.yaml
index 53f642275..5d0bbada5 100644
--- a/continuous_integration/travis/travis-38.yaml
+++ b/continuous_integration/travis/travis-38.yaml
@@ -35,6 +35,7 @@ dependencies:
   - crick
   - cytoolz
   - distributed==2.0.1
+  - msgpack-python==0.6.2
   - graphviz
   - ipython
   - lz4
diff --git a/dask/array/core.py b/dask/array/core.py
index ff59dedef..0054e6307 100644
--- a/dask/array/core.py
+++ b/dask/array/core.py
@@ -667,7 +667,7 @@ def map_blocks(
                         (
                             cached_cumsum(arg.chunks[j], initial_zero=True)
                             if ind in out_ind
-                            else np.array([0, arg.shape[j]])
+                            else [0, arg.shape[j]]
                         )
                         for j, ind in enumerate(in_ind)
                     ]
@@ -1126,7 +1126,7 @@ class Array(DaskMethodsMixin):
 
     @property
     def shape(self):
-        return tuple(map(sum, self.chunks))
+        return tuple(cached_cumsum(c, initial_zero=True)[-1] for c in self.chunks)
 
     @property
     def chunksize(self):
diff --git a/dask/array/creation.py b/dask/array/creation.py
index f42e2b336..963056c7a 100644
--- a/dask/array/creation.py
+++ b/dask/array/creation.py
@@ -579,8 +579,8 @@ def diagonal(a, offset=0, axis1=0, axis2=1):
 
     diag_chunks = []
     chunk_offsets = []
-    cum1 = list(cached_cumsum(a.chunks[axis1], initial_zero=True)[:-1])
-    cum2 = list(cached_cumsum(a.chunks[axis2], initial_zero=True)[:-1])
+    cum1 = cached_cumsum(a.chunks[axis1], initial_zero=True)[:-1]
+    cum2 = cached_cumsum(a.chunks[axis2], initial_zero=True)[:-1]
     for co1, c1 in zip(cum1, a.chunks[axis1]):
         chunk_offsets.append([])
         for co2, c2 in zip(cum2, a.chunks[axis2]):
diff --git a/dask/array/random.py b/dask/array/random.py
index 881a9c6ec..b43e2434f 100644
--- a/dask/array/random.py
+++ b/dask/array/random.py
@@ -1,4 +1,5 @@
 import numbers
+import warnings
 from itertools import product
 from numbers import Integral
 from operator import getitem
@@ -16,7 +17,22 @@ from .core import (
 from .creation import arange
 from ..base import tokenize
 from ..highlevelgraph import HighLevelGraph
-from ..utils import ignoring, random_state_data, derived_from
+from ..utils import ignoring, random_state_data, derived_from, skip_doctest
+
+
+def doc_wraps(func):
+    """ Copy docstring from one function to another """
+    warnings.warn(
+        "dask.array.random.doc_wraps is deprecated and will be removed in a future version",
+        FutureWarning,
+    )
+
+    def _(func2):
+        if func.__doc__ is not None:
+            func2.__doc__ = skip_doctest(func.__doc__)
+        return func2
+
+    return _
 
 
 class RandomState(object):
diff --git a/dask/array/slicing.py b/dask/array/slicing.py
index c20fdeca6..48d893ddd 100644
--- a/dask/array/slicing.py
+++ b/dask/array/slicing.py
@@ -1,12 +1,13 @@
 from itertools import product
 import math
 from numbers import Integral, Number
-from operator import getitem, itemgetter
+from operator import add, getitem, itemgetter
 import warnings
 import functools
+import bisect
 
 import numpy as np
-from toolz import memoize, merge, pluck, concat
+from toolz import memoize, merge, pluck, concat, accumulate
 
 from .. import core
 from ..highlevelgraph import HighLevelGraph
@@ -400,7 +401,7 @@ def _slice_1d(dim_shape, lengths, index):
 
     if isinstance(index, Integral):
         # use right-side search to be consistent with previous result
-        i = chunk_boundaries.searchsorted(index, side="right")
+        i = bisect.bisect_right(chunk_boundaries, index)
         if i > 0:
             # the very first chunk has no relative shift
             ind = index - chunk_boundaries[i - 1]
@@ -430,8 +431,8 @@ def _slice_1d(dim_shape, lengths, index):
 
     d = dict()
     if step > 0:
-        istart = chunk_boundaries.searchsorted(start, side="right")
-        istop = chunk_boundaries.searchsorted(stop, side="left")
+        istart = bisect.bisect_right(chunk_boundaries, start)
+        istop = bisect.bisect_left(chunk_boundaries, stop)
 
         # the bound is not exactly tight; make it tighter?
         istop = min(istop + 1, len(lengths))
@@ -452,8 +453,8 @@ def _slice_1d(dim_shape, lengths, index):
     else:
         rstart = start  # running start
 
-        istart = chunk_boundaries.searchsorted(start, side="left")
-        istop = chunk_boundaries.searchsorted(stop, side="right")
+        istart = bisect.bisect_left(chunk_boundaries, start)
+        istop = bisect.bisect_right(chunk_boundaries, stop)
 
         # the bound is not exactly tight; make it tighter?
         istart = min(istart + 1, len(chunk_boundaries) - 1)
@@ -1201,26 +1202,22 @@ class _HashIdWrapper(object):
 
 
 @functools.lru_cache()
-def _cumsum(seq):
+def _cumsum(seq, initial_zero):
     if isinstance(seq, _HashIdWrapper):
-        return _cumsum(seq.wrapped)
-    seq = np.array(seq)
-    dtype = np.int64 if np.issubdtype(seq.dtype, np.integer) else seq.dtype
-    out = np.empty(len(seq) + 1, dtype)
-    out[0] = 0
-    np.cumsum(seq, out=out[1:], dtype=dtype)
-    return out
+        seq = seq.wrapped
+    if initial_zero:
+        return tuple(accumulate(add, seq, 0))
+    else:
+        return tuple(accumulate(add, seq))
 
 
 def cached_cumsum(seq, initial_zero=False):
-    """Compute :meth:`np.cumsum` with caching.
+    """Compute :meth:`toolz.accumulate` with caching.
 
     Caching is by the identify of `seq` rather than the value. It is thus
     important that `seq` is a tuple of immutable objects, and this function
-    is intended for use where `seq` is a value that will persist.
-
-    The result has type int64 if the sequence contains integers, and
-    otherwise the type of ``np.array(seq)``.
+    is intended for use where `seq` is a value that will persist (generally
+    block sizes).
 
     Parameters
     ----------
@@ -1228,15 +1225,16 @@ def cached_cumsum(seq, initial_zero=False):
         Values to cumulatively sum.
     initial_zero : bool, optional
         If true, the return value is prefixed with a zero.
+
+    Returns
+    -------
+    tuple
     """
     if isinstance(seq, tuple):
         # Look up by identity first, to avoid a linear-time __hash__
         # if we've seen this tuple object before.
-        result = _cumsum(_HashIdWrapper(seq))
+        result = _cumsum(_HashIdWrapper(seq), initial_zero)
     else:
         # Construct a temporary tuple, and look up by value.
-        result = _cumsum(tuple(seq))
-
-    if not initial_zero:
-        result = result[1:]
+        result = _cumsum(tuple(seq), initial_zero)
     return result
diff --git a/dask/dataframe/methods.py b/dask/dataframe/methods.py
index 59c8523ae..9923739b9 100644
--- a/dask/dataframe/methods.py
+++ b/dask/dataframe/methods.py
@@ -257,8 +257,14 @@ def cummax_aggregate(x, y):
 
 
 def assign(df, *pairs):
-    kwargs = dict(partition(2, pairs))
-    return df.assign(**kwargs)
+    # Only deep copy when updating an element
+    # (to avoid modifying the original)
+    pairs = dict(partition(2, pairs))
+    deep = bool(set(pairs) & set(df.columns))
+    df = df.copy(deep=bool(deep))
+    for name, val in pairs.items():
+        df[name] = val
+    return df
 
 
 def unique(x, series_name=None):
diff --git a/docs/source/changelog.rst b/docs/source/changelog.rst
index 58fd9b949..f9c5066be 100644
--- a/docs/source/changelog.rst
+++ b/docs/source/changelog.rst
@@ -1,6 +1,66 @@
 Changelog
 =========
 
+2.11.0 / 2020-02-19
+-------------------
+
+Array
++++++
+
+- Cache result of ``Array.shape`` (:pr:`5916`) `Bruce Merry`_
+- Improve accuracy of ``estimate_graph_size`` for ``rechunk`` (:pr:`5907`) `Bruce Merry`_
+- Skip rechunk steps that do not alter chunking (:pr:`5909`) `Bruce Merry`_
+- Support ``dtype`` and other ``kwargs`` in ``coarsen`` (:pr:`5903`) `Matthew Rocklin`_
+- Push chunk override from ``map_blocks`` into blockwise (:pr:`5895`) `Bruce Merry`_
+- Avoid using ``rewrite_blockwise`` for a singleton (:pr:`5890`) `Bruce Merry`_
+- Optimize ``slices_from_chunks`` (:pr:`5891`) `Bruce Merry`_
+- Avoid unnecessary ``__getitem__`` in ``block()`` when chunks have correct dimensionality (:pr:`5884`) `Thomas Robitaille`_
+
+Bag
++++
+
+- Add ``include_path`` option for ``dask.bag.read_text`` (:pr:`5836`) `Yifan Gu`_
+- Fixes ``ValueError`` in delayed execution of bagged NumPy array (:pr:`5828`) `Surya Avala`_
+
+Core
+++++
+
+- CI: Pin ``msgpack`` (:pr:`5923`) `Tom Augspurger`_
+- Rename ``test_inner`` to ``test_outer`` (:pr:`5922`) `Shiva Raisinghani`_
+- ``quote`` should quote dicts too (:pr:`5905`) `Bruce Merry`_
+- Register a normalizer for literal (:pr:`5898`) `Bruce Merry`_
+- Improve layer name synthesis for non-HLGs (:pr:`5888`) `Bruce Merry`_
+- Replace flake8 pre-commit-hook with upstream (:pr:`5892`) `Julia Signell`_
+- Call pip as a module to avoid warnings (:pr:`5861`) `Cyril Shcherbin`_
+- Close ``ThreadPool`` at exit (:pr:`5852`) `Tom Augspurger`_
+- Remove ``dask.dataframe`` import in tokenization code (:pr:`5855`) `James Bourbeau`_
+
+DataFrame
++++++++++
+
+- Require ``pandas>=0.23`` (:pr:`5883`) `Tom Augspurger`_
+- Remove lambda from dataframe aggregation (:pr:`5901`) `Matthew Rocklin`_
+- Fix exception chaining in ``dataframe/__init__.py`` (:pr:`5882`) `Ram Rachum`_
+- Add support for reductions on empty dataframes (:pr:`5804`) `Shiva Raisinghani`_
+- Expose ``sort=`` argument for groupby (:pr:`5801`) `Richard J Zamora`_
+- Add ``df.empty`` property (:pr:`5711`) `rockwellw`_
+- Use parquet read speed-ups from ``fastparquet.api.paths_to_cats``. (:pr:`5821`) `Igor Gotlibovych`_
+
+Documentation
++++++++++++++
+
+- Deprecate ``doc_wraps`` (:pr:`5912`) `Tom Augspurger`_
+- Update array internal design docs for HighLevelGraph era (:pr:`5889`) `Bruce Merry`_
+- Move over dashboard connection docs (:pr:`5877`) `Matthew Rocklin`_
+- Move prometheus docs from distributed.dask.org (:pr:`5876`) `Matthew Rocklin`_
+- Removing duplicated DO block at the end (:pr:`5878`) `K.-Michael Aye`_
+- ``map_blocks`` see also (:pr:`5874`) `Tom Augspurger`_
+- More derived from (:pr:`5871`) `Julia Signell`_
+- Fix typo (:pr:`5866`) `Yetunde Dada`_
+- Fix typo in ``cloud.rst`` (:pr:`5860`) `Andrew Thomas`_
+- Add note pointing to code of conduct and diversity statement (:pr:`5844`) `Matthew Rocklin`_
+
+
 2.10.1 / 2020-01-30
 -------------------
 
@@ -2875,3 +2935,13 @@ Other
 .. _`Chris Roat`: https://github.com/ChrisRoat
 .. _`H. Thomson Comer`: https://github.com/thomcom
 .. _`Gerrit Holl`: https://github.com/gerritholl
+.. _`Thomas Robitaille`: https://github.com/astrofrog
+.. _`Yifan Gu`: https://github.com/gyf304
+.. _`Surya Avala`: https://github.com/suryaavala
+.. _`Cyril Shcherbin`: https://github.com/shcherbin
+.. _`Ram Rachum`: https://github.com/cool-RR
+.. _`Igor Gotlibovych`: https://github.com/ig248
+.. _`K.-Michael Aye`: https://github.com/michaelaye
+.. _`Yetunde Dada`: https://github.com/yetudada
+.. _`Andrew Thomas`: https://github.com/amcnicho
+.. _`rockwellw`: https://github.com/rockwellw


Test patch:
diff --git a/dask/array/tests/test_random.py b/dask/array/tests/test_random.py
index 3bcb4c397..bcf9bf491 100644
--- a/dask/array/tests/test_random.py
+++ b/dask/array/tests/test_random.py
@@ -356,3 +356,11 @@ def test_randint_dtype():
     assert_eq(x, x)
     assert x.dtype == "uint8"
     assert x.compute().dtype == "uint8"
+
+
+def test_doc_wraps_deprecated():
+    with pytest.warns(FutureWarning):
+
+        @da.random.doc_wraps(np.random.normal)
+        def f():
+            pass
diff --git a/dask/array/tests/test_routines.py b/dask/array/tests/test_routines.py
index a716fabf2..952c082f7 100644
--- a/dask/array/tests/test_routines.py
+++ b/dask/array/tests/test_routines.py
@@ -339,7 +339,7 @@ def test_vdot(shape, chunks):
 
 
 @pytest.mark.parametrize("shape1, shape2", [((20,), (6,)), ((4, 5), (2, 3))])
-def test_inner(shape1, shape2):
+def test_outer(shape1, shape2):
     np.random.random(1337)
 
     x = 2 * np.random.random(shape1) - 1
diff --git a/dask/array/tests/test_slicing.py b/dask/array/tests/test_slicing.py
index 378b6d35e..0be1eecdb 100644
--- a/dask/array/tests/test_slicing.py
+++ b/dask/array/tests/test_slicing.py
@@ -874,27 +874,23 @@ def test_cached_cumsum():
     a = (1, 2, 3, 4)
     x = cached_cumsum(a)
     y = cached_cumsum(a, initial_zero=True)
-    np.testing.assert_array_equal(x, [1, 3, 6, 10])
-    assert x.dtype == np.int64
-    np.testing.assert_array_equal(y, [0, 1, 3, 6, 10])
-    assert y.dtype == np.int64
+    assert x == (1, 3, 6, 10)
+    assert y == (0, 1, 3, 6, 10)
 
 
 def test_cached_cumsum_nan():
     a = (1, np.nan, 3)
     x = cached_cumsum(a)
     y = cached_cumsum(a, initial_zero=True)
-    np.testing.assert_array_equal(x, [1, np.nan, np.nan])
-    assert x.dtype == np.float64
-    np.testing.assert_array_equal(y, [0, 1, np.nan, np.nan])
-    assert y.dtype == np.float64
+    np.testing.assert_equal(x, (1, np.nan, np.nan))
+    np.testing.assert_equal(y, (0, 1, np.nan, np.nan))
 
 
 def test_cached_cumsum_non_tuple():
     a = [1, 2, 3]
-    np.testing.assert_array_equal(cached_cumsum(a), [1, 3, 6])
+    assert cached_cumsum(a) == (1, 3, 6)
     a[1] = 4
-    np.testing.assert_array_equal(cached_cumsum(a), [1, 5, 8])
+    assert cached_cumsum(a) == (1, 5, 8)
 
 
 @pytest.mark.parametrize("params", [(2, 2, 1), (5, 3, 2)])