Repository: dask/dask

Issue:
Issue with trim_excess in dask.array.coarsen
I am trying to use ``dask.array.coarsen`` on 3-d arrays and am finding that the result depends on the chunking of the data, which doesn't seem right. The following example illustrates the issue:

```python
In [1]: import numpy as np                                                      

In [2]: import dask.array as da                                                 

In [3]: array1 = da.random.random((623, 768, 768)).rechunk((322, 256, 256))     

In [4]: array2 = da.random.random((623, 768, 768)).rechunk((623, 768, 768))     

In [5]: import numpy as np                                                      

In [6]: da.coarsen(np.nanmean, array1, {1: 3}, trim_excess=True).shape          
Out[6]: (623, 255, 768)

In [7]: da.coarsen(np.nanmean, array2, {1: 3}, trim_excess=True).shape          
Out[7]: (623, 256, 768)
```

Since 768 is divisible by 3, I would expect the final array to have shape ``(623, 256, 768)`` as for ``array2``. The result for ``array1`` seems incorrect, so I think this is a bug?


Discussion:

Patch:
diff --git a/dask/array/core.py b/dask/array/core.py
index b2583bbae..a47254ece 100644
--- a/dask/array/core.py
+++ b/dask/array/core.py
@@ -3903,7 +3903,7 @@ def elemwise(op, *args, **kwargs):
     for arg in args:
         shape = getattr(arg, "shape", ())
         if any(is_dask_collection(x) for x in shape):
-            # Want to excluded Delayed shapes and dd.Scalar
+            # Want to exclude Delayed shapes and dd.Scalar
             shape = ()
         shapes.append(shape)
 
diff --git a/dask/array/reshape.py b/dask/array/reshape.py
index 18cef220e..ac99db2e7 100644
--- a/dask/array/reshape.py
+++ b/dask/array/reshape.py
@@ -159,7 +159,7 @@ def reshape(x, shape):
     shape = tuple(map(sanitize_index, shape))
     known_sizes = [s for s in shape if s != -1]
     if len(known_sizes) < len(shape):
-        if len(known_sizes) - len(shape) > 1:
+        if len(shape) - len(known_sizes) > 1:
             raise ValueError("can only specify one unknown dimension")
         # Fastpath for x.reshape(-1) on 1D arrays, allows unknown shape in x
         # for this case only.
diff --git a/dask/array/routines.py b/dask/array/routines.py
index 8c3ebce2d..d57ae6c09 100644
--- a/dask/array/routines.py
+++ b/dask/array/routines.py
@@ -5,6 +5,7 @@ from collections.abc import Iterable
 from functools import wraps, partial
 from numbers import Real, Integral
 from distutils.version import LooseVersion
+from typing import List
 
 import numpy as np
 from tlz import concat, sliding_window, interleave
@@ -1403,17 +1404,76 @@ def piecewise(x, condlist, funclist, *args, **kw):
     )
 
 
+def aligned_coarsen_chunks(chunks: List[int], multiple: int) -> List[int]:
+    """
+    Returns a new chunking aligned with the coarsening multiple.
+    Any excess is at the end of the array.
+
+    Examples
+    --------
+    >>> aligned_coarsen_chunks(chunks=(1, 2, 3), multiple=4)
+    (4, 2)
+    >>> aligned_coarsen_chunks(chunks=(1, 20, 3, 4), multiple=4)
+    (20, 4, 4)
+    >>> aligned_coarsen_chunks(chunks=(20, 10, 15, 23, 24), multiple=10)
+    (20, 10, 20, 20, 20, 2)
+    """
+
+    def choose_new_size(multiple, q, left):
+        """
+        See if multiple * q is a good choice when 'left' elements are remaining.
+        Else return multiple * (q-1)
+        """
+        possible = multiple * q
+        if (left - possible) > 0:
+            return possible
+        else:
+            return multiple * (q - 1)
+
+    newchunks = []
+    left = sum(chunks) - sum(newchunks)
+    chunkgen = (c for c in chunks)
+    while left > 0:
+        if left < multiple:
+            newchunks.append(left)
+            break
+
+        chunk_size = next(chunkgen, 0)
+        if chunk_size == 0:
+            chunk_size = multiple
+
+        q, r = divmod(chunk_size, multiple)
+        if q == 0:
+            continue
+        elif r == 0:
+            newchunks.append(chunk_size)
+        elif r >= 5:
+            newchunks.append(choose_new_size(multiple, q + 1, left))
+        else:
+            newchunks.append(choose_new_size(multiple, q, left))
+
+        left = sum(chunks) - sum(newchunks)
+
+    return tuple(newchunks)
+
+
 @wraps(chunk.coarsen)
 def coarsen(reduction, x, axes, trim_excess=False, **kwargs):
-    if not trim_excess and not all(
-        bd % div == 0 for i, div in axes.items() for bd in x.chunks[i]
-    ):
+    if not trim_excess and not all(x.shape[i] % div == 0 for i, div in axes.items()):
         msg = "Coarsening factor does not align with block dimensions"
         raise ValueError(msg)
 
     if "dask" in inspect.getfile(reduction):
         reduction = getattr(np, reduction.__name__)
 
+    new_chunks = {}
+    for i, div in axes.items():
+        aligned = aligned_coarsen_chunks(x.chunks[i], div)
+        if aligned != x.chunks[i]:
+            new_chunks[i] = aligned
+    if new_chunks:
+        x = x.rechunk(new_chunks)
+
     name = "coarsen-" + tokenize(reduction, x, axes, trim_excess)
     dsk = {
         (name,)
diff --git a/dask/dataframe/core.py b/dask/dataframe/core.py
index c93a153b3..60c3bab6b 100644
--- a/dask/dataframe/core.py
+++ b/dask/dataframe/core.py
@@ -1141,8 +1141,8 @@ Dask Name: {name}, {task} tasks"""
             ``'12h'`` or ``pd.Timedelta(hours=12)``.  Assumes a datetime index.
         force : bool, default False
             Allows the expansion of the existing divisions.
-            If False then the new divisions lower and upper bounds must be
-            the same as the old divisions.
+            If False then the new divisions' lower and upper bounds must be
+            the same as the old divisions'.
 
         Notes
         -----
@@ -2801,7 +2801,7 @@ Dask Name: {name}, {task} tasks""".format(
             from the input series and the transformation. Ignored for
             non-callable/dict-like ``index`` or when the input series has
             unknown divisions. Note that this may only be set to ``True`` if
-            you know that the transformed index is monotonicly increasing. Dask
+            you know that the transformed index is monotonically increasing. Dask
             will check that transformed divisions are monotonic, but cannot
             check all the values between divisions, so incorrectly setting this
             can result in bugs.
@@ -3535,7 +3535,10 @@ class DataFrame(_Frame):
             # Slicing with integer labels is always iloc based except for a
             # float indexer for some reason
             if is_integer_slice and not is_float_dtype(self.index.dtype):
-                self.iloc[key]
+                # NOTE: this always fails currently, as iloc is mostly
+                # unsupported, but we call it anyway here for future-proofing
+                # and error-attribution purposes
+                return self.iloc[key]
             else:
                 return self.loc[key]
 
@@ -5336,7 +5339,7 @@ def _rename_dask(df, names):
     Destructively rename columns of dd.DataFrame or name of dd.Series.
     Not for pd.DataFrame or pd.Series.
 
-    Internaly used to overwrite dd.DataFrame.columns and dd.Series.name
+    Internally used to overwrite dd.DataFrame.columns and dd.Series.name
     We can't use map_partition because it applies function then rename
 
     Parameters
diff --git a/dask/dataframe/multi.py b/dask/dataframe/multi.py
index becd0e001..ca635fa4b 100644
--- a/dask/dataframe/multi.py
+++ b/dask/dataframe/multi.py
@@ -11,7 +11,7 @@ There are two important cases:
 2.  We combine along an unpartitioned index or other column
 
 In the first case we know which partitions of each dataframe interact with
-which others.  This lets uss be significantly more clever and efficient.
+which others.  This lets us be significantly more clever and efficient.
 
 In the second case each partition from one dataset interacts with all
 partitions from the other.  We handle this through a shuffle operation.


Test patch:
diff --git a/dask/array/tests/test_array_core.py b/dask/array/tests/test_array_core.py
index ec65561d2..197159398 100644
--- a/dask/array/tests/test_array_core.py
+++ b/dask/array/tests/test_array_core.py
@@ -4138,7 +4138,7 @@ def test_dask_array_holds_scipy_sparse_containers():
 
     z = x.T.map_blocks(scipy.sparse.csr_matrix)
     zz = z.compute(scheduler="single-threaded")
-    assert isinstance(yy, scipy.sparse.spmatrix)
+    assert isinstance(zz, scipy.sparse.spmatrix)
     assert (zz == xx.T).all()
 
 
diff --git a/dask/array/tests/test_reshape.py b/dask/array/tests/test_reshape.py
index c453f28a6..3bdceb86a 100644
--- a/dask/array/tests/test_reshape.py
+++ b/dask/array/tests/test_reshape.py
@@ -1,6 +1,8 @@
 import pytest
 import numpy as np
+import dask.array as da
 from dask.array.reshape import reshape_rechunk, expand_tuple, contract_tuple
+from dask.array.utils import assert_eq
 
 
 @pytest.mark.parametrize(
@@ -61,3 +63,19 @@ def test_contract_tuple():
     assert contract_tuple((1, 1, 2, 5, 1), 2) == (2, 2, 4, 2)
     assert contract_tuple((2, 4), 2) == (2, 4)
     assert contract_tuple((2, 4), 3) == (6,)
+
+
+def test_reshape_unknown_sizes():
+    a = np.random.random((10, 6, 6))
+    A = da.from_array(a, chunks=(5, 2, 3))
+
+    a2 = a.reshape((60, -1))
+    A2 = A.reshape((60, -1))
+
+    assert A2.shape == (60, 6)
+    assert_eq(A2, a2)
+
+    with pytest.raises(ValueError):
+        a.reshape((60, -1, -1))
+    with pytest.raises(ValueError):
+        A.reshape((60, -1, -1))
diff --git a/dask/array/tests/test_routines.py b/dask/array/tests/test_routines.py
index b657d9098..532262bb9 100644
--- a/dask/array/tests/test_routines.py
+++ b/dask/array/tests/test_routines.py
@@ -1521,13 +1521,31 @@ def test_coarsen():
 
 def test_coarsen_with_excess():
     x = da.arange(10, chunks=5)
-    assert_eq(da.coarsen(np.min, x, {0: 3}, trim_excess=True), np.array([0, 5]))
+    assert_eq(da.coarsen(np.min, x, {0: 5}, trim_excess=True), np.array([0, 5]))
     assert_eq(
         da.coarsen(np.sum, x, {0: 3}, trim_excess=True),
-        np.array([0 + 1 + 2, 5 + 6 + 7]),
+        np.array([0 + 1 + 2, 3 + 4 + 5, 6 + 7 + 8]),
     )
 
 
+def test_coarsen_bad_chunks():
+
+    x1 = da.arange(10, chunks=5)
+    x2 = x1.rechunk((1, 2, 3, 4))
+    assert_eq(da.coarsen(np.sum, x1, {0: 5}), da.coarsen(np.sum, x2, {0: 5}))
+
+
+def test_aligned_coarsen_chunks():
+
+    from ..routines import aligned_coarsen_chunks as acc
+
+    assert acc((20, 10, 15, 23, 24), 10) == (20, 10, 20, 20, 20, 2)
+    assert acc((20, 10, 15, 42, 23, 24), 10) == (20, 10, 20, 40, 20, 20, 4)
+    assert acc((20, 10, 15, 47, 23, 24), 10) == (20, 10, 20, 50, 20, 10, 9)
+    assert acc((2, 10, 15, 47, 23, 24), 10) == (10, 20, 50, 20, 20, 1)
+    assert acc((10, 20, 30, 40, 2), 10) == (10, 20, 30, 40, 2)
+
+
 def test_insert():
     x = np.random.randint(10, size=(10, 10))
     a = da.from_array(x, chunks=(5, 5))
diff --git a/dask/dataframe/tests/test_dataframe.py b/dask/dataframe/tests/test_dataframe.py
index 28a0beb22..8b2aca252 100644
--- a/dask/dataframe/tests/test_dataframe.py
+++ b/dask/dataframe/tests/test_dataframe.py
@@ -2609,7 +2609,7 @@ def test_to_dask_array_raises(as_frame):
         a.to_dask_array(5)
 
 
-@pytest.mark.parametrize("as_frame", [False, False])
+@pytest.mark.parametrize("as_frame", [False, True])
 def test_to_dask_array_unknown(as_frame):
     s = pd.Series([1, 2, 3, 4, 5], name="foo")
     a = dd.from_pandas(s, chunksize=2)
@@ -2622,11 +2622,12 @@ def test_to_dask_array_unknown(as_frame):
     result = result.chunks
 
     if as_frame:
+        assert len(result) == 2
         assert result[1] == (1,)
+    else:
+        assert len(result) == 1
 
-    assert len(result) == 1
     result = result[0]
-
     assert len(result) == 2
     assert all(np.isnan(x) for x in result)