Repository: dask/dask

Issue:
`2024.08.0` array slicing does not preserve masks
<!-- Please include a self-contained copy-pastable example that generates the issue if possible.

Please be concise with code posted. See guidelines below on how to provide a good bug report:

- Craft Minimal Bug Reports http://matthewrocklin.com/blog/work/2018/02/28/minimal-bug-reports
- Minimal Complete Verifiable Examples https://stackoverflow.com/help/mcve

Bug reports that follow these guidelines are easier to diagnose, and so are often handled much more quickly.
-->

**Describe the issue**:

Since dask 2024.8.0, slicing a masked array does not preserve the mask.

**Minimal Complete Verifiable Example**:

```python
import numpy as np
import dask.array as da

arr = np.ma.array(range(8), mask=[0, 0, 1, 0, 0, 1, 0, 1])
darr = da.from_array(arr, chunks=(4, 4))

print(arr[[2, 6]])
print(darr[[2, 6]].compute())
```
Output with 2024.8.0:
```
[-- 6]
[2 6]
```
This works as expected  with 2024.7.0:
```
[-- 6]
[-- 6]
```


**Anything else we need to know?**:

I suspect this line may be at least part of the problem
https://github.com/dask/dask/blob/1ba9c5bf9d0b005b89a04804f29c16e5f0f2dadf/dask/array/_shuffle.py#L190

`np.concatenate` does not preserve masks, although there is an open PR to change that: https://github.com/numpy/numpy/pull/22913.  For masked arrays there is [np.ma.concatenate](https://numpy.org/doc/stable/reference/generated/numpy.ma.concatenate.html#numpy-ma-concatenate).

I have not been able to test any further because my internet connection is currently _flakey_ and my attempt to make a dev environment failed.

**Environment**:

- Dask version: 2024.8.0
- Python version: 3.12.5
- Operating System: Ubuntu
- Install method (conda, pip, source): conda


Discussion:
github-actions[bot]: ## Unit Test Results
_See [test report](https://dask.github.io/dask/test_report.html) for an extended history of previous test failures. This is useful for diagnosing flaky tests._

     13 files  ± 0       13 suites  ±0   2h 57m 33s :stopwatch: + 1m 5s
 13 132 tests + 1   12 064 :white_check_mark: + 1   1 068 :zzz: ±0  0 :x: ±0 
136 838 runs  +11  117 772 :white_check_mark: +11  19 066 :zzz: ±0  0 :x: ±0 

Results for commit df7981fc. ± Comparison against base commit 1ba9c5bf.

[test-results]:data:application/gzip;base64,H4sIABLMuWYC/03Myw6DIBCF4VcxrLuYQWWgL9PILSH10iCsTN+9WhVd/t9kzsJ86N3MnhXWj4rNOaSrbI5dCtO4NYh2o/Wa9jvW/OzXnI3ZkINoLnyHz/4pi/ku9KtBARfjFA+Jedynhazl0WUaiYgXPKYVCHHabfrf92UzDUNIazDrSUn0prXOAEenO60bgSC90E6TBaWQnCf2/QFuvGgbGgEAAA==


Patch:
diff --git a/dask/array/_shuffle.py b/dask/array/_shuffle.py
index 37aacb4bb..89f435d7b 100644
--- a/dask/array/_shuffle.py
+++ b/dask/array/_shuffle.py
@@ -8,6 +8,7 @@ import numpy as np
 from dask import config
 from dask.array.chunk import getitem
 from dask.array.core import Array, unknown_chunk_message
+from dask.array.dispatch import concatenate_lookup
 from dask.base import tokenize
 from dask.highlevelgraph import HighLevelGraph
 
@@ -187,7 +188,8 @@ def _shuffle(chunks, indexer, axis, in_name, out_name, token):
 
 
 def concatenate_arrays(arrs, sorter, axis):
-    return np.take(np.concatenate(arrs, axis=axis), np.argsort(sorter), axis=axis)
+    concatenate = concatenate_lookup.dispatch(type(arrs[0]))
+    return np.take(concatenate(arrs, axis=axis), np.argsort(sorter), axis=axis)
 
 
 def convert_key(key, chunk, axis):
diff --git a/dask/array/core.py b/dask/array/core.py
index b710fe491..3009dcb11 100644
--- a/dask/array/core.py
+++ b/dask/array/core.py
@@ -4524,6 +4524,13 @@ def retrieve_from_ooc(
     return load_dsk
 
 
+def _as_dtype(a, dtype):
+    if dtype is None:
+        return a
+    else:
+        return a.astype(dtype)
+
+
 def asarray(
     a, allow_unknown_chunksizes=False, dtype=None, order=None, *, like=None, **kwargs
 ):
@@ -4573,16 +4580,22 @@ def asarray(
     >>> y = [[1, 2, 3], [4, 5, 6]]
     >>> da.asarray(y)
     dask.array<array, shape=(2, 3), dtype=int64, chunksize=(2, 3), chunktype=numpy.ndarray>
+
+    .. warning::
+        `order` is ignored if `a` is an `Array`, has the attribute ``to_dask_array``,
+        or is a list or tuple of `Array`'s.
     """
     if like is None:
         if isinstance(a, Array):
-            return a
+            return _as_dtype(a, dtype)
         elif hasattr(a, "to_dask_array"):
-            return a.to_dask_array()
+            return _as_dtype(a.to_dask_array(), dtype)
         elif type(a).__module__.split(".")[0] == "xarray" and hasattr(a, "data"):
-            return asarray(a.data)
+            return _as_dtype(asarray(a.data, order=order), dtype)
         elif isinstance(a, (list, tuple)) and any(isinstance(i, Array) for i in a):
-            return stack(a, allow_unknown_chunksizes=allow_unknown_chunksizes)
+            return _as_dtype(
+                stack(a, allow_unknown_chunksizes=allow_unknown_chunksizes), dtype
+            )
         elif not isinstance(getattr(a, "shape", None), Iterable):
             a = np.asarray(a, dtype=dtype, order=order)
     else:
@@ -4641,16 +4654,20 @@ def asanyarray(a, dtype=None, order=None, *, like=None, inline_array=False):
     >>> y = [[1, 2, 3], [4, 5, 6]]
     >>> da.asanyarray(y)
     dask.array<array, shape=(2, 3), dtype=int64, chunksize=(2, 3), chunktype=numpy.ndarray>
+
+    .. warning::
+        `order` is ignored if `a` is an `Array`, has the attribute ``to_dask_array``,
+        or is a list or tuple of `Array`'s.
     """
     if like is None:
         if isinstance(a, Array):
-            return a
+            return _as_dtype(a, dtype)
         elif hasattr(a, "to_dask_array"):
-            return a.to_dask_array()
+            return _as_dtype(a.to_dask_array(), dtype)
         elif type(a).__module__.split(".")[0] == "xarray" and hasattr(a, "data"):
-            return asanyarray(a.data)
+            return _as_dtype(asarray(a.data, order=order), dtype)
         elif isinstance(a, (list, tuple)) and any(isinstance(i, Array) for i in a):
-            return stack(a)
+            return _as_dtype(stack(a), dtype)
         elif not isinstance(getattr(a, "shape", None), Iterable):
             a = np.asanyarray(a, dtype=dtype, order=order)
     else:


Test patch:
diff --git a/dask/array/tests/test_array_core.py b/dask/array/tests/test_array_core.py
index 70eeb0a9f..d975cfced 100644
--- a/dask/array/tests/test_array_core.py
+++ b/dask/array/tests/test_array_core.py
@@ -2704,6 +2704,18 @@ def test_asarray(asarray):
     assert_eq(asarray(y), x)
 
 
+@pytest.mark.parametrize("asarray", [da.asarray, da.asanyarray])
+def test_asarray_array_dtype(asarray):
+    # test array input
+    x = asarray([1, 2])
+    assert_eq(asarray(x, dtype=da.float32), np.asarray(x, dtype=np.float32))
+
+    x = asarray(x, dtype=da.float64)
+    assert x.dtype == da.float64
+    x = asarray(x, dtype=da.int32)
+    assert x.dtype == da.int32
+
+
 @pytest.mark.parametrize("asarray", [da.asarray, da.asanyarray])
 def test_asarray_dask_dataframe(asarray):
     # https://github.com/dask/dask/issues/3885
diff --git a/dask/array/tests/test_slicing.py b/dask/array/tests/test_slicing.py
index ed8a99e0c..1bde2a01e 100644
--- a/dask/array/tests/test_slicing.py
+++ b/dask/array/tests/test_slicing.py
@@ -1069,6 +1069,12 @@ def test_slice_array_3d_with_bool_numpy_array():
     assert_eq(actual, expected)
 
 
+def test_slice_masked_arrays():
+    arr = np.ma.array(range(8), mask=[0, 0, 1, 0, 0, 1, 0, 1])
+    darr = da.from_array(arr, chunks=(4, 4))
+    assert_eq(darr[[2, 6]], arr[[2, 6]])
+
+
 def test_slice_array_null_dimension():
     array = da.from_array(np.zeros((3, 0)))
     expected = np.zeros((3, 0))[[0]]