Repository: dask/dask

Issue:
Error when using a Numpy boolean mask on a Dask array
I am trying to extract values from a Dask array using a boolean Numpy array and am running into the following error:

```python
In [8]: import numpy as np                                                                                                                                                                                                                                                                                                                                                                                                                                                                              

In [9]: from dask.array import random                                                                                                                                                                                                                                                                                                                                                                                                                                                                   

In [10]: array = random.random((4, 3, 2))                                                                                                                                                                                                                                                                                                                                                                                                                                                               

In [11]: mask = np.random.random((4, 3, 2)) > 0.5                                                                                                                                                                                                                                                                                                                                                                                                                                                       

In [12]: array[mask]                                                                                                                                                                                                                                                                                                                                                                                                                                                                                    
---------------------------------------------------------------------------
IndexError                                Traceback (most recent call last)
<ipython-input-12-38297b4b7b0f> in <module>
----> 1 array[mask]

~/python/dev/lib/python3.8/site-packages/dask/array/core.py in __getitem__(self, index)
   1537         )
   1538 
-> 1539         index2 = normalize_index(index, self.shape)
   1540 
   1541         dependencies = {self.name}

~/python/dev/lib/python3.8/site-packages/dask/array/slicing.py in normalize_index(idx, shape)
    813     for i, d in zip(idx, none_shape):
    814         if d is not None:
--> 815             check_index(i, d)
    816     idx = tuple(map(sanitize_index, idx))
    817     idx = tuple(map(normalize_slice, idx, none_shape))

~/python/dev/lib/python3.8/site-packages/dask/array/slicing.py in check_index(ind, dimension)
    867         if x.dtype == bool:
    868             if x.size != dimension:
--> 869                 raise IndexError(
    870                     "Boolean array length %s doesn't equal dimension %s"
    871                     % (x.size, dimension)

IndexError: Boolean array length 24 doesn't equal dimension 4
```

Note that converting the mask to a dask array fixes this:

```python
In [14]: from dask.array import asarray                                                                                                                                                                                                                                                                                                                                                                                                                                                                 

In [15]: array[asarray(mask)]                                                                                                                                                                                                                                                                                                                                                                                                                                                                           
Out[15]: dask.array<getitem, shape=(nan,), dtype=float64, chunksize=(nan,), chunktype=numpy.ndarray>

In [16]: array[asarray(mask)].compute()                                                                                                                                                                                                                                                                                                                                                                                                                                                                 
Out[16]: 
array([0.87541677, 0.72561386, 0.80415762, 0.90781457, 0.41807167,
       0.88501242, 0.33921153, 0.5392456 , 0.31133289, 0.86372707,
       0.32839977, 0.77310795])
```

but I thought I should report this as a bug.

Discussion:
jsignell: @dask/maintenance for a review. 

Patch:
diff --git a/dask/array/core.py b/dask/array/core.py
index ef8970091..a5c51ca87 100644
--- a/dask/array/core.py
+++ b/dask/array/core.py
@@ -4269,13 +4269,11 @@ def stack(seq, axis=0, allow_unknown_chunksizes=False):
     if not seq:
         raise ValueError("Need array(s) to stack")
     if not allow_unknown_chunksizes and not all(x.shape == seq[0].shape for x in seq):
-        idx = np.where(np.asanyarray([x.shape for x in seq]) != seq[0].shape)[0]
+        idx = first(i for i in enumerate(seq) if i[1].shape != seq[0].shape)
         raise ValueError(
             "Stacked arrays must have the same shape. "
-            "The first {0} had shape {1}, while array "
-            "{2} has shape {3}".format(
-                idx[0], seq[0].shape, idx[0] + 1, seq[idx[0]].shape
-            )
+            "The first array had shape {0}, while array "
+            "{1} has shape {2}.".format(seq[0].shape, idx[0] + 1, idx[1].shape)
         )
 
     meta = np.stack([meta_from_array(a) for a in seq], axis=axis)
diff --git a/dask/array/creation.py b/dask/array/creation.py
index 01668df6c..6e7bd92e5 100644
--- a/dask/array/creation.py
+++ b/dask/array/creation.py
@@ -27,7 +27,7 @@ from .wrap import empty, ones, zeros, full
 from .utils import AxisError, meta_from_array, zeros_like_safe
 
 
-def empty_like(a, dtype=None, chunks=None):
+def empty_like(a, dtype=None, order="C", chunks=None, name=None):
     """
     Return a new array with the same shape and type as a given array.
 
@@ -38,9 +38,15 @@ def empty_like(a, dtype=None, chunks=None):
         returned array.
     dtype : data-type, optional
         Overrides the data type of the result.
+    order : {'C', 'F'}, optional
+        Whether to store multidimensional data in C- or Fortran-contiguous
+        (row- or column-wise) order in memory.
     chunks : sequence of ints
         The number of samples on each block. Note that the last block will have
         fewer samples if ``len(array) % chunks != 0``.
+    name : str, optional
+        An optional keyname for the array. Defaults to hashing the input
+        keyword arguments.
 
     Returns
     -------
@@ -67,11 +73,13 @@ def empty_like(a, dtype=None, chunks=None):
     return empty(
         a.shape,
         dtype=(dtype or a.dtype),
+        order=order,
         chunks=(chunks if chunks is not None else a.chunks),
+        name=name,
     )
 
 
-def ones_like(a, dtype=None, chunks=None):
+def ones_like(a, dtype=None, order="C", chunks=None, name=None):
     """
     Return an array of ones with the same shape and type as a given array.
 
@@ -82,9 +90,15 @@ def ones_like(a, dtype=None, chunks=None):
         the returned array.
     dtype : data-type, optional
         Overrides the data type of the result.
+    order : {'C', 'F'}, optional
+        Whether to store multidimensional data in C- or Fortran-contiguous
+        (row- or column-wise) order in memory.
     chunks : sequence of ints
         The number of samples on each block. Note that the last block will have
         fewer samples if ``len(array) % chunks != 0``.
+    name : str, optional
+        An optional keyname for the array. Defaults to hashing the input
+        keyword arguments.
 
     Returns
     -------
@@ -104,11 +118,13 @@ def ones_like(a, dtype=None, chunks=None):
     return ones(
         a.shape,
         dtype=(dtype or a.dtype),
+        order=order,
         chunks=(chunks if chunks is not None else a.chunks),
+        name=name,
     )
 
 
-def zeros_like(a, dtype=None, chunks=None):
+def zeros_like(a, dtype=None, order="C", chunks=None, name=None):
     """
     Return an array of zeros with the same shape and type as a given array.
 
@@ -119,9 +135,15 @@ def zeros_like(a, dtype=None, chunks=None):
         the returned array.
     dtype : data-type, optional
         Overrides the data type of the result.
+    order : {'C', 'F'}, optional
+        Whether to store multidimensional data in C- or Fortran-contiguous
+        (row- or column-wise) order in memory.
     chunks : sequence of ints
         The number of samples on each block. Note that the last block will have
         fewer samples if ``len(array) % chunks != 0``.
+    name : str, optional
+        An optional keyname for the array. Defaults to hashing the input
+        keyword arguments.
 
     Returns
     -------
@@ -141,11 +163,13 @@ def zeros_like(a, dtype=None, chunks=None):
     return zeros(
         a.shape,
         dtype=(dtype or a.dtype),
+        order=order,
         chunks=(chunks if chunks is not None else a.chunks),
+        name=name,
     )
 
 
-def full_like(a, fill_value, dtype=None, chunks=None):
+def full_like(a, fill_value, order="C", dtype=None, chunks=None, name=None):
     """
     Return a full array with the same shape and type as a given array.
 
@@ -158,9 +182,15 @@ def full_like(a, fill_value, dtype=None, chunks=None):
         Fill value.
     dtype : data-type, optional
         Overrides the data type of the result.
+    order : {'C', 'F'}, optional
+        Whether to store multidimensional data in C- or Fortran-contiguous
+        (row- or column-wise) order in memory.
     chunks : sequence of ints
         The number of samples on each block. Note that the last block will have
         fewer samples if ``len(array) % chunks != 0``.
+    name : str, optional
+        An optional keyname for the array. Defaults to hashing the input
+        keyword arguments.
 
     Returns
     -------
@@ -182,8 +212,10 @@ def full_like(a, fill_value, dtype=None, chunks=None):
     return full(
         a.shape,
         fill_value,
+        order=order,
         dtype=(dtype or a.dtype),
         chunks=(chunks if chunks is not None else a.chunks),
+        name=name,
     )
 
 
diff --git a/dask/array/slicing.py b/dask/array/slicing.py
index ffd5bc7c6..25c7ee448 100644
--- a/dask/array/slicing.py
+++ b/dask/array/slicing.py
@@ -769,9 +769,10 @@ def normalize_index(idx, shape):
     1.  Replaces ellipses with many full slices
     2.  Adds full slices to end of index
     3.  Checks bounding conditions
-    4.  Replaces numpy arrays with lists
-    5.  Posify's integers and lists
-    6.  Normalizes slices to canonical form
+    4.  Replace multidimensional numpy arrays with dask arrays
+    5.  Replaces numpy arrays with lists
+    6.  Posify's integers and lists
+    7.  Normalizes slices to canonical form
 
     Examples
     --------
@@ -785,9 +786,20 @@ def normalize_index(idx, shape):
     (slice(7, None, None),)
     >>> normalize_index((Ellipsis, None), (10,))
     (slice(None, None, None), None)
+    >>> normalize_index(np.array([[True, False], [False, True], [True, True]]), (3, 2))
+    (dask.array<array, shape=(3, 2), dtype=bool, chunksize=(3, 2), chunktype=numpy.ndarray>,)
     """
+    from .core import from_array
+
     if not isinstance(idx, tuple):
         idx = (idx,)
+
+    # if a > 1D numpy.array is provided, cast it to a dask array
+    if len(idx) > 0 and len(shape) > 1:
+        i = idx[0]
+        if isinstance(i, np.ndarray) and i.shape == shape:
+            idx = (from_array(i), *idx[1:])
+
     idx = replace_ellipsis(len(shape), idx)
     n_sliced_dims = 0
     for i in idx:
diff --git a/dask/array/wrap.py b/dask/array/wrap.py
index 82b8e8005..af3d1487f 100644
--- a/dask/array/wrap.py
+++ b/dask/array/wrap.py
@@ -18,6 +18,7 @@ def _parse_wrap_args(func, args, kwargs, shape):
     if not isinstance(shape, (tuple, list)):
         shape = (shape,)
 
+    name = kwargs.pop("name", None)
     chunks = kwargs.pop("chunks", "auto")
 
     dtype = kwargs.pop("dtype", None)
@@ -26,7 +27,6 @@ def _parse_wrap_args(func, args, kwargs, shape):
     dtype = np.dtype(dtype)
 
     chunks = normalize_chunks(chunks, shape, dtype=dtype)
-    name = kwargs.pop("name", None)
 
     name = name or funcname(func) + "-" + tokenize(
         func, shape, chunks, dtype, args, kwargs
@@ -121,8 +121,8 @@ def wrap(wrap_func, func, **kwargs):
     template = """
     Blocked variant of %(name)s
 
-    Follows the signature of %(name)s exactly except that it also requires a
-    keyword argument chunks=(...)
+    Follows the signature of %(name)s exactly except that it also features
+    optional keyword arguments ``chunks: int, tuple, or dict`` and ``name: str``.
 
     Original signature follows below.
     """
@@ -137,13 +137,39 @@ w = wrap(wrap_func_shape_as_first_arg)
 ones = w(np.ones, dtype="f8")
 zeros = w(np.zeros, dtype="f8")
 empty = w(np.empty, dtype="f8")
-full = w(np.full)
 
 
 w_like = wrap(wrap_func_like_safe)
 
 
-ones_like = w_like(np.ones, func_like=np.ones_like)
-zeros_like = w_like(np.zeros, func_like=np.zeros_like)
 empty_like = w_like(np.empty, func_like=np.empty_like)
-full_like = w_like(np.full, func_like=np.full_like)
+
+
+# full and full_like require special casing due to argument check on fill_value
+# Generate wrapped functions only once
+_full = w(np.full)
+_full_like = w_like(np.full, func_like=np.full_like)
+
+
+def full(shape, fill_value, *args, **kwargs):
+    # np.isscalar has somewhat strange behavior:
+    # https://docs.scipy.org/doc/numpy/reference/generated/numpy.isscalar.html
+    if np.ndim(fill_value) != 0:
+        raise ValueError(
+            f"fill_value must be scalar. Received {type(fill_value).__name__} instead."
+        )
+    return _full(shape=shape, fill_value=fill_value, *args, **kwargs)
+
+
+def full_like(a, fill_value, *args, **kwargs):
+    if np.ndim(fill_value) != 0:
+        raise ValueError(
+            f"fill_value must be scalar. Received {type(fill_value).__name__} instead."
+        )
+    return _full_like(a=a, fill_value=fill_value, *args, **kwargs,)
+
+
+full.__doc__ = _full.__doc__
+full.__name__ = _full.__name__
+full_like.__doc__ = _full_like.__doc__
+full_like.__name__ = _full_like.__name__
diff --git a/dask/dataframe/tseries/resample.py b/dask/dataframe/tseries/resample.py
index 851ff2a92..502126020 100644
--- a/dask/dataframe/tseries/resample.py
+++ b/dask/dataframe/tseries/resample.py
@@ -124,16 +124,7 @@ class Resampler(object):
             )
             raise ValueError(msg)
         self.obj = obj
-        rule = pd.tseries.frequencies.to_offset(rule)
-        day_nanos = pd.tseries.frequencies.Day().nanos
-
-        if getnanos(rule) and day_nanos % rule.nanos:
-            raise NotImplementedError(
-                "Resampling frequency %s that does"
-                " not evenly divide a day is not "
-                "implemented" % rule
-            )
-        self._rule = rule
+        self._rule = pd.tseries.frequencies.to_offset(rule)
         self._kwargs = kwargs
 
     def _agg(self, how, meta=None, fill_value=np.nan, how_args=(), how_kwargs={}):


Test patch:
diff --git a/dask/array/tests/test_array_core.py b/dask/array/tests/test_array_core.py
index 765e14ebb..d8241de0f 100644
--- a/dask/array/tests/test_array_core.py
+++ b/dask/array/tests/test_array_core.py
@@ -2314,8 +2314,7 @@ def test_from_array_tasks_always_call_getter(x, chunks):
 
 
 def test_from_array_ndarray_onechunk():
-    """ndarray with a single chunk produces a minimal single key dict
-    """
+    """ndarray with a single chunk produces a minimal single key dict"""
     x = np.array([[1, 2], [3, 4]])
     dx = da.from_array(x, chunks=-1)
     assert_eq(x, dx)
@@ -2334,8 +2333,7 @@ def test_from_array_ndarray_getitem():
 
 @pytest.mark.parametrize("x", [[1, 2], (1, 2), memoryview(b"abc")])
 def test_from_array_list(x):
-    """Lists, tuples, and memoryviews are automatically converted to ndarray
-    """
+    """Lists, tuples, and memoryviews are automatically converted to ndarray"""
     dx = da.from_array(x, chunks=-1)
     assert_eq(np.array(x), dx)
     assert isinstance(dx.dask[dx.name, 0], np.ndarray)
@@ -2348,8 +2346,7 @@ def test_from_array_list(x):
 
 @pytest.mark.parametrize("type_", [t for t in np.ScalarType if t is not memoryview])
 def test_from_array_scalar(type_):
-    """Python and numpy scalars are automatically converted to ndarray
-    """
+    """Python and numpy scalars are automatically converted to ndarray"""
     if type_ == np.datetime64:
         x = np.datetime64("2000-01-01")
     else:
@@ -3621,8 +3618,10 @@ def test_stack_errs():
     with pytest.raises(ValueError) as e:
         da.stack([da.zeros((2,), chunks=2)] * 10 + [da.zeros((3,), chunks=3)] * 10)
 
-    assert "shape" in str(e.value).lower()
-    assert "(2,)" in str(e.value)
+    assert (
+        str(e.value)
+        == "Stacked arrays must have the same shape. The first array had shape (2,), while array 11 has shape (3,)."
+    )
     assert len(str(e.value)) < 105
 
 
diff --git a/dask/array/tests/test_creation.py b/dask/array/tests/test_creation.py
index b63b34d8c..1928c5fe8 100644
--- a/dask/array/tests/test_creation.py
+++ b/dask/array/tests/test_creation.py
@@ -29,8 +29,10 @@ from dask.array.numpy_compat import _numpy_117, _numpy_118
 @pytest.mark.parametrize("cast_shape", [tuple, list, np.asarray])
 @pytest.mark.parametrize("cast_chunks", [tuple, list, np.asarray])
 @pytest.mark.parametrize("shape, chunks", [((10, 10), (4, 4))])
+@pytest.mark.parametrize("name", [None, "my-name"])
+@pytest.mark.parametrize("order", ["C", "F"])
 @pytest.mark.parametrize("dtype", ["i4"])
-def test_arr_like(funcname, shape, cast_shape, dtype, cast_chunks, chunks):
+def test_arr_like(funcname, shape, cast_shape, dtype, cast_chunks, chunks, name, order):
     np_func = getattr(np, funcname)
     da_func = getattr(da, funcname)
     shape = cast_shape(shape)
@@ -48,11 +50,11 @@ def test_arr_like(funcname, shape, cast_shape, dtype, cast_chunks, chunks):
     if "like" in funcname:
         a = np.random.randint(0, 10, shape).astype(dtype)
 
-        np_r = np_func(a)
-        da_r = da_func(a, chunks=chunks)
+        np_r = np_func(a, order=order)
+        da_r = da_func(a, order=order, chunks=chunks, name=name)
     else:
-        np_r = np_func(shape, dtype=dtype)
-        da_r = da_func(shape, dtype=dtype, chunks=chunks)
+        np_r = np_func(shape, order=order, dtype=dtype)
+        da_r = da_func(shape, order=order, dtype=dtype, chunks=chunks, name=name)
 
     assert np_r.shape == da_r.shape
     assert np_r.dtype == da_r.dtype
@@ -60,6 +62,16 @@ def test_arr_like(funcname, shape, cast_shape, dtype, cast_chunks, chunks):
     if "empty" not in funcname:
         assert (np_r == np.asarray(da_r)).all()
 
+    if name is None:
+        assert funcname.split("_")[0] in da_r.name
+    else:
+        assert da_r.name == name
+
+    if "order" == "F":
+        assert np.isfortran(da_r.compute())
+    else:
+        assert not np.isfortran(da_r.compute())
+
 
 @pytest.mark.parametrize("endpoint", [True, False])
 def test_linspace(endpoint):
diff --git a/dask/array/tests/test_slicing.py b/dask/array/tests/test_slicing.py
index d457035a8..53429b8f4 100644
--- a/dask/array/tests/test_slicing.py
+++ b/dask/array/tests/test_slicing.py
@@ -948,3 +948,13 @@ def test_gh4043(lock, asarray, fancy):
     a2 = da.from_array(np.ones(3), chunks=1, asarray=asarray, lock=lock, fancy=fancy)
     al = da.stack([a1, a2])
     assert_eq(al, al)
+
+
+def test_slice_array_3d_with_bool_numpy_array():
+    # https://github.com/dask/dask/issues/6089
+    array = da.arange(0, 24).reshape((4, 3, 2))
+    mask = np.arange(0, 24).reshape((4, 3, 2)) > 12
+
+    actual = array[mask].compute()
+    expected = np.arange(13, 24)
+    assert_eq(actual, expected)
diff --git a/dask/array/tests/test_wrap.py b/dask/array/tests/test_wrap.py
index f3d9345c9..6de3943bf 100644
--- a/dask/array/tests/test_wrap.py
+++ b/dask/array/tests/test_wrap.py
@@ -42,6 +42,17 @@ def test_full():
     assert a.name.startswith("full-")
 
 
+def test_full_error_nonscalar_fill_value():
+    with pytest.raises(ValueError, match="fill_value must be scalar"):
+        da.full((3, 3), [100, 100], chunks=(2, 2), dtype="i8")
+
+
+def test_full_like_error_nonscalar_fill_value():
+    x = np.full((3, 3), 1, dtype="i8")
+    with pytest.raises(ValueError, match="fill_value must be scalar"):
+        da.full_like(x, [100, 100], chunks=(2, 2), dtype="i8")
+
+
 def test_can_make_really_big_array_of_ones():
     ones((1000000, 1000000), chunks=(100000, 100000))
     ones(shape=(1000000, 1000000), chunks=(100000, 100000))
diff --git a/dask/dataframe/tseries/tests/test_resample.py b/dask/dataframe/tseries/tests/test_resample.py
index c8a657fe4..327b43927 100644
--- a/dask/dataframe/tseries/tests/test_resample.py
+++ b/dask/dataframe/tseries/tests/test_resample.py
@@ -128,12 +128,32 @@ def test_resample_pads_last_division_to_avoid_off_by_one():
     assert_eq(actual, expected)
 
 
-def test_series_resample_not_implemented():
-    index = pd.date_range(start="2012-01-02", periods=100, freq="T")
+def test_resample_does_not_evenly_divide_day():
+    import numpy as np
+
+    index = pd.date_range("2012-01-02", "2012-02-02", freq="H")
+    index = index.union(pd.date_range("2012-03-02", "2012-04-02", freq="H"))
+    df = pd.DataFrame({"p": np.random.random(len(index))}, index=index)
+    ddf = dd.from_pandas(df, npartitions=5)
+    # Frequency doesn't evenly divide day
+    expected = df.resample("2D").count()
+    result = ddf.resample("2D").count().compute()
+
+    assert_eq(result, expected)
+
+
+def test_series_resample_does_not_evenly_divide_day():
+    index = pd.date_range("2012-01-02 00:00:00", "2012-01-02 01:00:00", freq="T")
+    index = index.union(
+        pd.date_range("2012-01-02 06:00:00", "2012-01-02 08:00:00", freq="T")
+    )
     s = pd.Series(range(len(index)), index=index)
     ds = dd.from_pandas(s, npartitions=5)
     # Frequency doesn't evenly divide day
-    pytest.raises(NotImplementedError, lambda: resample(ds, "57T"))
+    expected = s.resample("57T").mean()
+    result = ds.resample("57T").mean().compute()
+
+    assert_eq(result, expected)
 
 
 def test_unknown_divisions_error():